Redis 缓存设计模式

缓存是提升系统性能最有效的手段之一，但选择错误的模式或缺乏异常处理反而会引入不一致和稳定性问题。

Cache-Aside 旁路缓存

读时先查缓存，miss 再查 DB 并回填；写时先更新 DB 再删除缓存（推荐删除而非更新，避免并发脏读）。实现简单但需处理击穿穿透雪崩。

import redis, json, hashlib, random
from functools import wraps
from typing import Callable, Any

r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

def cache_aside(ttl: int = 300, jitter: float = 0.1):
    def decorator(fn: Callable):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            k = f"c:{fn.__name__}:{hashlib.md5(str((args,sorted(kwargs.items()))).encode()).hexdigest()[:10]}"
            cached = r.get(k)
            if cached == "__NULL__": return None
            if cached is not None: return json.loads(cached)
            result = fn(*args, **kwargs)
            t = int(ttl * (1 + random.uniform(-jitter, jitter)))
            if result is None: r.setex(k, max(30, t//5), "__NULL__")
            else: r.setex(k, max(1, t), json.dumps(result, default=str))
            return result
        return wrapper
    return decorator

def write_through(key: str, value: Any, db_update: Callable, ttl: int = 300):
    pipe = r.pipeline()
    pipe.setex(key, ttl, json.dumps(value, default=str))
    try:
        db_update(value)
        pipe.execute()
    except Exception:
        pipe.reset(); raise

bg_queue: list[tuple[str, Any]] = []
def write_behind(key: str, value: Any):
    bg_queue.append((key, value))
    r.setex(key, 60, json.dumps(value, default=str))

def flush_bg(db_update: Callable):
    while bg_queue:
        k, v = bg_queue.pop(0)
        try: db_update(k, v)
        except Exception: bg_queue.insert(0, (k, v)); break

@cache_aside(ttl=600)
def get_product(pid: int):
    return {"id": pid, "name": f"Product {pid}", "price": 99.99}

Write-Through / Write-Behind

Write-Through 写缓存同时同步写 DB，原子性强但延迟高。Write-Behind 先写缓存，异步批量刷 DB，吞吐高但存在短窗口数据丢失风险。

最佳实践

设置随机 TTL ±10% 防止雪崩，热点 Key 加互斥锁或逻辑过期防击穿，不存在的数据写空值防穿透。