Redis 缓存设计模式：穿透、击穿、雪崩解决方案

缓存三大问题

1. 缓存穿透

问题：查询不存在的数据，请求直接打到数据库。

解决方案：

# 方案 1：缓存空值
from redis import Redis
redis = Redis()

def get_user(user_id):
    data = redis.get(f"user:{user_id}")
    if data == "NULL":  # 缓存空值
        return None
    if data:
        return json.loads(data)
    
    # 查询数据库
    user = db.query(user_id)
    if not user:
        redis.setex(f"user:{user_id}", 60, "NULL")  # 空值缓存 60 秒
        return None
    
    redis.setex(f"user:{user_id}", 300, json.dumps(user))
    return user

# 方案 2：布隆过滤器
from pybloom_live import BloomFilter
bloom = BloomFilter(capacity=1000000, error_rate=0.001)

def get_user(user_id):
    if user_id not in bloom:  # 一定不存在
        return None
    # 继续查询缓存和数据库

2. 缓存击穿

问题：热点 key 过期，大量请求同时打到数据库。

解决方案：

import threading

lock = threading.Lock()

def get_hot_data(key):
    data = redis.get(key)
    if data:
        return json.loads(data)
    
    # 互斥锁
    with lock:
        # 双重检查
        data = redis.get(key)
        if data:
            return json.loads(data)
        
        # 查询数据库
        result = db.query(key)
        redis.setex(key, 300, json.dumps(result))
        return result

# 方案 2：永不过期 + 异步更新
def get_data_with_async_update(key):
    data = redis.get(key)
    if data:
        # 检查是否即将过期
        ttl = redis.ttl(key)
        if ttl < 60:  # 小于 60 秒
            # 异步更新
            threading.Thread(target=update_cache, args=(key,)).start()
        return json.loads(data)

3. 缓存雪崩

问题：大量 key 同时过期，数据库压力激增。

解决方案：

import random

def set_cache_with_jitter(key, value, base_ttl=300):
    # 添加随机过期时间（±10%）
    jitter = random.randint(-30, 30)
    ttl = base_ttl + jitter
    redis.setex(key, ttl, json.dumps(value))

# 方案 2：分层缓存
def get_data(key):
    # L1 本地缓存（5 分钟）
    data = local_cache.get(key)
    if data:
        return data
    
    # L2 Redis 缓存（30 分钟）
    data = redis.get(key)
    if data:
        local_cache.set(key, data, 300)
        return json.loads(data)
    
    # 查询数据库
    result = db.query(key)
    redis.setex(key, 1800, json.dumps(result))
    local_cache.set(key, result, 300)
    return result

最佳实践总结

问题	核心原因	解决方案
穿透	查询不存在的数据	缓存空值、布隆过滤器
击穿	热点 key 过期	互斥锁、异步更新
雪崩	大量 key 同时过期	随机过期时间、分层缓存

监控与告警

# 监控缓存命中率
def monitor_cache():
    info = redis.info("stats")
    hits = info["keyspace_hits"]
    misses = info["keyspace_misses"]
    hit_rate = hits / (hits + misses) if (hits + misses) > 0 else 0
    
    if hit_rate < 0.8:  # 命中率低于 80% 告警
        send_alert(f"缓存命中率过低：{hit_rate:.2%}")

合理设计缓存策略，让系统更稳定高效！