redis-patterns

Redis 模式

常见后端用例下 Redis 最佳实践快速参考。

工作原理

Redis 是一种内存数据结构存储，支持字符串、哈希、列表、集合、有序集合、流等。单个实例上的 Redis 命令是原子性的；多步骤工作流需要 Lua 脚本、MULTI/EXEC 事务或显式同步来保持原子性。数据可通过 RDB 快照或 AOF 日志选择持久化。客户端通过 TCP 使用 RESP 协议通信；连接池对于避免每个请求的握手开销至关重要。

何时激活

• 为应用程序添加缓存
• 实现限流或节流
• 构建分布式锁或协调
• 设置会话或令牌存储
• 使用 Pub/Sub 或 Redis Streams 进行消息传递
• 在生产环境中配置 Redis（池化、驱逐、集群）

数据结构速查表

用例	结构	示例键
简单缓存	字符串	product:123
用户会话	哈希	session:abc
排行榜	有序集合	scores:weekly
唯一访客	集合	visitors:2024-01-01
活动订阅源	列表	feed:user:456
事件流	流	events:orders
计数器/限流	字符串 (INCR)	ratelimit:user:123
布隆过滤器/HLL	HyperLogLog	hll:pageviews

核心模式

缓存旁路（延迟加载）

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

写入穿透缓存

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))

缓存失效

# 基于标签的失效——将相关键分组到一个集合下
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)

会话存储

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-py 库，它实现了完整的 Redlock 算法。

Pub/Sub 与流

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)  # 限制流长度

# 消费者组 — 保证至少一次投递
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)

当需要投递保证、消费者组或回放时，优先选择流而非 Pub/Sub。

键设计

命名约定

# 模式: resource:id:field
user:123:profile
order:456:status
cache:product:789

# 模式: namespace:resource:id
myapp:session:abc123
myapp:ratelimit:user:123

# 模式: resource:date（时间限制键）
stats:pageviews:2024-01-01

TTL 策略

数据类型	建议 TTL
用户会话	24小时 (86400)
API 响应缓存	5–15 分钟
限流窗口	匹配窗口大小
短期令牌	5–10 分钟
排行榜	1小时–24小时
静态/参考数据	1小时–1周

始终设置 TTL。没有 TTL 的键会无限累积，导致内存压力。

连接管理

连接池

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)

集群模式

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	驱逐最近最少使用	通用缓存
volatile-lru	仅在带 TTL 的键中执行 LRU	混合数据存储
allkeys-lfu	驱逐使用频率最低	偏斜访问模式
volatile-ttl	驱逐即将过期的	优先保留长生命周期数据

通过 redis.conf 设置：maxmemory-policy allkeys-lru

反模式

反模式	问题	修复
无 TTL 的键	内存无限制增长	始终设置 TTL
生产环境使用 KEYS *	阻塞服务器 (O(N))	使用 SCAN 游标
存储大对象 (>100KB)	序列化缓慢，内存压力	存储引用并从对象存储获取
用单一 Redis 做所有事	缓存与队列之间无隔离	使用单独的 DB 或实例
忽略连接池限制	负载下连接耗尽	根据工作负载调整池大小
未处理缓存未命中雪崩	冷启动时惊群效应	使用锁或概率性提前过期
不经思考的 FLUSHALL	清除整个实例	按键模式限制删除范围

缓存未命中雪崩预防

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)  # 获取锁后重新检查
        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。键基于请求参数。

按用户限流 API： 对于低流量端点，使用固定窗口配合 pipeline(transaction=True)；对于高精度按用户限流，使用滑动窗口 Lua。

跨工作线程协调后台任务： 使用 acquire_lock，并将 TTL 设置为超过预期的任务持续时间。始终在 finally 块中释放锁。

向多个订阅者扇形分发通知： 对于即发即弃，使用 Pub/Sub。如果需要保证投递或为延迟消费者提供回放，则切换到 Streams。

快速参考

模式	何时使用
缓存旁路	读密集，能容忍轻微数据滞后
写入穿透	需要强一致性
分布式锁	防止对资源的并发访问
滑动窗口限流	高精度按用户限流
Redis Streams	具有消费者组的持久事件队列
Pub/Sub	不需要投递保证的广播
有序集合排行榜	排名评分、分页
HyperLogLog	低内存下的近似唯一计数

相关

• 技能：postgres-patterns — 关系型数据模式
• 技能：backend-patterns — API 和服务层模式
• 技能：database-migrations — 模式版本管理
• 技能：django-patterns — Django 缓存框架集成
• 代理：database-reviewer — 完整数据库审查工作流