redis-patterns

Redis Patterns

常见后端业务场景下的 Redis 最佳实践速查指南。

工作原理

Redis 是一款基于内存的数据结构存储系统，支持字符串（String）、哈希（Hash）、列表（List）、集合（Set）、有序集合（Sorted Set）、流（Stream）等多种数据类型。在单机实例上，Redis 的单条命令都是原子性的；如果是多步骤的工作流，则需要使用 Lua 脚本、MULTI/EXEC 事务或显式同步机制来保证原子性。Redis 支持通过 RDB 快照或 AOF 日志进行数据持久化。客户端通过 RESP 协议基于 TCP 与 Redis 通信；在实际生产环境中，使用连接池至关重要，能有效避免每次请求建立连接带来的开销。

触发条件

• 给应用添加缓存层
• 实现限流、请求防刷或削峰
• 构建分布式锁或分布式协调机制
• 实现 Session 会话或 Token 存储
• 使用 Pub/Sub 或 Redis Streams 处理消息传递
• 生产环境配置 Redis（连接池、内存淘汰策略、集群等）

数据结构速查表

适用场景	数据结构	示例 Key
简单缓存	String	product:123
用户 Session	Hash	session:abc
排行榜	Sorted Set	scores:weekly
独立访客统计 (UV)	Set	visitors:2024-01-01
活动/动态 Feed 流	List	feed:user:456
事件流/消息队列	Stream	events:orders
计数器 / 限流	String (INCR)	ratelimit:user:123
布隆过滤器 / 基数统计	HyperLogLog	hll:pageviews

核心模式

旁路缓存 (Cache-Aside / 懒加载)

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_product(product_id: int):
    cache_key = f"product:{product_id}"
    cached = r.get(cache_key)

    if cached:
        return json.loads(cached)

    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    r.setex(cache_key, 3600, json.dumps(product))  # TTL: 1小时
    return product

直写缓存 (Write-Through Cache)

def update_product(product_id: int, data: dict):
    # 先写入数据库
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)

    # 立即更新缓存
    cache_key = f"product:{product_id}"
    r.setex(cache_key, 3600, json.dumps(data))

缓存失效策略

# 基于 Tag 的缓存失效 —— 将关联的 key 归类到一个 Set 下
def cache_product(product_id: int, category_id: int, data: dict):
    key = f"product:{product_id}"
    tag = f"tag:category:{category_id}"
    pipe = r.pipeline(transaction=True)
    pipe.setex(key, 3600, json.dumps(data))
    pipe.sadd(tag, key)
    pipe.expire(tag, 3600)
    pipe.execute()

def invalidate_category(category_id: int):
    tag = f"tag:category:{category_id}"
    keys = r.smembers(tag)
    if keys:
        r.delete(*keys)
    r.delete(tag)

Session 存储

import time
import uuid

def create_session(user_id: int, ttl: int = 86400) -> str:
    session_id = str(uuid.uuid4())
    key = f"session:{session_id}"
    pipe = r.pipeline(transaction=True)
    pipe.hset(key, mapping={
        "user_id": user_id,
        "created_at": int(time.time()),
    })
    pipe.expire(key, ttl)
    pipe.execute()
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.hgetall(f"session:{session_id}")
    return data if data else None

def delete_session(session_id: str):
    r.delete(f"session:{session_id}")

限流

固定窗口算法 (简单限流)

def is_rate_limited(user_id: int, limit: int = 100, window: int = 60) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window}"
    pipe = r.pipeline(transaction=True)
    pipe.incr(key)
    pipe.expire(key, window)
    count, _ = pipe.execute()
    return count > limit

滑动窗口算法 (Lua 脚本 — 原子操作)

-- sliding_window.lua
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)

if count < limit then
    -- 使用唯一成员（当前时间戳 + 序列号）防止同一毫秒内发生碰撞冲突
    local seq_key = key .. ':seq'
    local seq = redis.call('INCR', seq_key)
    redis.call('EXPIRE', seq_key, math.ceil(window / 1000))
    redis.call('ZADD', key, now, now .. '-' .. seq)
    redis.call('EXPIRE', key, math.ceil(window / 1000))
    return 1
end
return 0

sliding_window = r.register_script(open('sliding_window.lua').read())

def allow_request(user_id: int) -> bool:
    key = f"ratelimit:sliding:{user_id}"
    now = int(time.time() * 1000)
    return bool(sliding_window(keys=[key], args=[now, 60000, 100]))

分布式锁

单节点分布式锁 (SET NX PX)

import uuid

def acquire_lock(resource: str, ttl_ms: int = 5000) -> str | None:
    lock_key = f"lock:{resource}"
    token = str(uuid.uuid4())
    acquired = r.set(lock_key, token, px=ttl_ms, nx=True)
    return token if acquired else None

def release_lock(resource: str, token: str) -> bool:
    release_script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    result = r.eval(release_script, 1, f"lock:{resource}", token)
    return bool(result)

# 使用示例
token = acquire_lock("order:payment:123")
if token:
    try:
        process_payment()
    finally:
        release_lock("order:payment:123", token)

提示：如果是多节点集群架构，建议使用实现了完整 Redlock 算法的 redlock-py 库。

发布/订阅与消息流

Pub/Sub (即发即弃)

# 发布者
def publish_event(channel: str, payload: dict):
    r.publish(channel, json.dumps(payload))

# 订阅者（阻塞式 — 请在独立线程/进程中运行）
def subscribe_events(channel: str):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            handle(json.loads(message['data']))

Redis Streams (持久化队列)

# 生产者
def emit(stream: str, event: dict):
    r.xadd(stream, event, maxlen=10000)  # 限制 Stream 最大长度

# 消费者组 —— 保证至少一次（at-least-once）投递
try:
    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)
except Exception:
    pass  # 消费者组已存在

def consume(stream: str, group: str, consumer: str):
    while True:
        messages = r.xreadgroup(group, consumer, {stream: '>'}, count=10, block=2000)
        for _, entries in (messages or []):
            for msg_id, data in entries:
                process(data)
                r.xack(stream, group, msg_id)

提示：在需要可靠投递保证、消费者组或消息重放功能时，优先使用 Streams 而非 Pub/Sub。

Key 设计规范

命名约定

# 格式：资源名:ID:字段名
user:123:profile
order:456:status
cache:product:789

# 格式：命名空间:资源名:ID
myapp:session:abc123
myapp:ratelimit:user:123

# 格式：资源名:日期（按时间维度划分的 key）
stats:pageviews:2024-01-01

TTL 过期策略建议

数据类型	建议 TTL
用户 Session	24h (86400)
API 响应缓存	5–15 分钟
限流时间窗口	与限流窗口时长保持一致
短时 Token	5–10 分钟
排行榜数据	1h–24h
静态/基础引用数据	1h–1 周

务必设置 TTL。没有设置 TTL 的 key 会一直堆积，导致内存膨胀压力过大。

连接管理

连接池 (Connection Pooling)

from redis import ConnectionPool, Redis

pool = ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=20,
    decode_responses=True,
    socket_connect_timeout=2,
    socket_timeout=2,
)

r = Redis(connection_pool=pool)

集群模式 (Cluster Mode)

from redis.cluster import RedisCluster

r = RedisCluster(
    startup_nodes=[{"host": "redis-1", "port": 6379}],
    decode_responses=True,
    skip_full_coverage_check=True,
)

哨兵模式 (Sentinel - 高可用)

from redis.sentinel import Sentinel

sentinel = Sentinel(
    [('sentinel-1', 26379), ('sentinel-2', 26379)],
    socket_timeout=0.5,
)
master = sentinel.master_for('mymaster', decode_responses=True)
replica = sentinel.slave_for('mymaster', decode_responses=True)

内存淘汰策略

策略	淘汰行为	最适用场景
noeviction	内存满时写入报错	队列 / 核心关键数据
allkeys-lru	淘汰最近最少使用的 key	通用缓存
volatile-lru	仅在设置了 TTL 的 key 中淘汰 LRU	混合存储场景
allkeys-lfu	淘汰最不经常使用的 key	访问热点极度倾斜的场景
volatile-ttl	淘汰最快到期的 key	优先保留长 TTL 数据

在 redis.conf 中配置：maxmemory-policy allkeys-lru

反模式与踩坑避雷

反模式 / 坏习惯	带来的问题	正确做法
不给 Key 设置 TTL	内存无限制增长直到撑爆	始终显式设置 TTL
生产环境直接运行 KEYS *	阻塞 Redis 线程（O(N) 复杂度）	改用游标分步迭代 SCAN
存储超大 Blob（>100KB）	序列化变慢、占用极大内存	仅存资源引用地址，大文件存对象存储
所有业务共用单台 Redis 实例	缓存与队列相互挤占资源，缺乏隔离	按业务拆分 DB 或使用独立实例
忽视连接池上限配置	高并发下连接耗尽引发超时死锁	根据业务吞吐量合理配置连接池大小
不处理缓存击穿/防刷雪崩	冷启动或热 Key 失效致数据库被冲垮	使用互斥锁或概率性提前续期
乱用 FLUSHALL 命令	瞬间清空整台实例所有数据	根据 key 前缀匹配进行精细化删除

预防缓存击穿 (Cache Miss Stampede)

import threading

_locks: dict[str, threading.Lock] = {}
_locks_mutex = threading.Lock()

def get_with_lock(key: str, fetch_fn, ttl: int = 300):
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    with _locks_mutex:
        if key not in _locks:
            _locks[key] = threading.Lock()
        lock = _locks[key]
    with lock:
        cached = r.get(key)  # 抢到锁后二次检查（Double-check）
        if cached:
            return json.loads(cached)
        value = fetch_fn()
        r.setex(key, ttl, json.dumps(value))
        return value

注意：如果是多进程部署环境，请将进程内锁替换为上方“分布式锁”章节提供的 acquire_lock/release_lock。

实战示例

为 Django/Flask API 接口添加缓存：
采用 Cache-Aside（旁路缓存）模式，使用 setex 对接口响应设置 5 分钟 TTL，基于请求参数组合 Key。

针对单用户做 API 接口限流：
对于低频接口可以使用固定窗口算法配合 pipeline(transaction=True)；对于精度要求高的限流场景，建议使用滑动窗口 Lua 脚本。

多 Worker 跨进程/节点协调后台任务：
使用 acquire_lock 加锁，设置超过预计任务执行时间的 TTL，且务必在 finally 块中释放锁。

向多个订阅者广播通知：
即发即弃的广播场景直接用 Pub/Sub；如果需要保证消息不丢失或离线消费重放，切到 Redis Streams。

常用速查

模式/方案	何时使用
旁路缓存 (Cache-aside)	读多写少，能容忍短时间数据不一致
直写缓存 (Write-through)	对数据一致性要求较高
分布式锁 (Distributed lock)	防止多节点并发竞争同一资源
滑动窗口限流 (Sliding window rate limit)	精准的按用户/接口频率削峰限流
Redis Streams	需要消费者组支持的高可用持久化消息队列
发布/订阅 (Pub/Sub)	广播通知，无需投递保证或消息留存
Sorted Set 排行榜	带计分的实时排名与分页查询
HyperLogLog	低内存消耗的大规模基数估算（如 UV）

关联 Skill

• Skill: postgres-patterns — 关系型数据库模式
• Skill: backend-patterns — API 与服务层设计模式
• Skill: database-migrations — 数据库 Schema 版本管理
• Skill: django-patterns — Django 缓存框架集成
• Agent: database-reviewer — 数据库完整评审工作流