Redis 缓存击穿、穿透与雪崩：从原理到可落地的三层防御实战

一、一个真实的故障场景

某电商商品详情页平时缓存命中率在 95% 以上，数据库 QPS 只有几十。一次大促前，一个爆款商品的缓存刚好过期，同一秒内上千个请求同时发现缓存为空，全部打到数据库，主库 CPU 瞬间打满，接口超时又引发重试，雪上加霜。事后复盘发现，这不是单纯的“缓存没命中”，而是击穿、穿透、雪崩三类问题经常同时出现，需要分层防御。

二、三者先分清，别用一种办法治所有病

判断方法也很直接：看监控。如果是某一个 Key 的回源量暴涨，多半是击穿；如果是大量不存在的 ID 在刷库，多半是穿透；如果是整体缓存命中率断崖式下跌、回源量全量上涨，多半是雪崩。

三、第一层：防击穿——互斥锁 + 逻辑过期

1. 互斥锁：只放一个请求去重建缓存

核心是“先抢锁再查库”，抢不到锁的请求短暂等待后重试读缓存，而不是直接查库。锁一定要带过期时间，持有者标识要用随机值，释放时用 Lua 脚本比对标识，防止误删别人的锁。

import time, uuid
import redis

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
UNLOCK_SCRIPT = """
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end
"""

def get_product(pid):
    key = f'product:{pid}'
    data = r.get(key)
    if data:
        return data
    lock_key = f'lock:{key}'
    token = str(uuid.uuid4())
    # 只抢 10 秒锁，避免重建失败导致死锁
    if r.set(lock_key, token, nx=True, px=10000):
        try:
            data = r.get(key)  # 抢到锁后先复查，避免重复查库
            if data:
                return data
            data = query_db(pid)  # 真实查库
            r.setex(key, 3600, data)
            return data
        finally:
            r.eval(UNLOCK_SCRIPT, 1, lock_key, token)
    # 未抢到锁：等 50 毫秒后重试读缓存，最多重试 20 次
    for _ in range(20):
        time.sleep(0.05)
        data = r.get(key)
        if data:
            return data
    return query_db(pid)  # 兜底，避免无限等待

注意：锁超时时间要大于一次重建耗时；如果重建本身很慢，宁可先返回旧数据或降级，也不要把锁时间设得无限长。

2. 逻辑过期：热点 Key 永不过期，过期后异步重建

把真实过期时间写进 Value 里，Redis 的 Key 本身设置一个较长的 TTL 或不过期。读到逻辑过期的数据时，先返回旧数据，再由拿到锁的线程异步重建。这样热点流量永远不会在同一时刻全部穿透。

{
  "data": {"id": 1001, "price": 299, "stock": 120},
  "expire_at": 1761696000
}

这个方法适合“可以短暂返回旧数据”的场景，例如商品详情、榜单、配置类数据；库存、价格强一致的场景不要用逻辑过期，必须走互斥锁加实时校验。

四、第二层：防穿透——先校验，再给空值一个短缓存

1. 参数校验与布隆过滤器

对明显非法的 ID、超长参数直接拦截，不进缓存也不进数据库。对合法格式但可能不存在的 ID，可以用布隆过滤器先挡一层：把全量有效 ID 预热进过滤器，请求先查过滤器，不存在就直接返回空。布隆过滤器有误判率，但不会漏判真实存在的数据，所以误判只会导致少量请求多查一次库，不会丢数据。

2. 空值也要缓存，但 TTL 要短

数据库查不到时，不要直接返回而不写缓存，否则攻击者用同一个不存在的 ID 可以无限刷库。正确做法是写一个特殊空值占位，并设置 30 到 120 秒的短 TTL，业务层读到占位符就返回空。

def get_user(uid):
    if uid <= 0:
        return None
    key = f'user:{uid}'
    val = r.get(key)
    if val == '__NULL__':
        return None
    if val:
        return val
    row = query_db(uid)
    if row is None:
        r.setex(key, 60, '__NULL__')  # 空值短缓存，防止反复穿透
        return None
    r.setex(key, 3600, row)
    return row

注意：空值缓存和正常缓存要用同一个 Key 命名空间管理，否则缓存清理、预热时容易遗漏；同时要监控空值命中率，突然暴涨往往意味着有人在刷不存在的 ID。

五、第三层：防雪崩——打散过期时间 + 多级缓存 + 限流熔断

1. TTL 加随机抖动，不要整点过期

最常见的雪崩就是所有 Key 都写了同样的 3600 秒，结果一小时后同时失效。改为基础 TTL 加随机抖动：

import random
base_ttl = 3600
ttl = base_ttl + random.randint(0, 600)
r.setex(key, ttl, data)

批量预热、定时任务刷新缓存时也要分批、分时段做，不要在同一分钟内全量刷新。

2. 本地缓存做一级缓冲

在应用内加一层小容量本地缓存，例如 Caffeine 或 Guava Cache，TTL 设为 30 到 60 秒。即使 Redis 短暂不可用，本地缓存也能挡住大部分读流量，给 Redis 恢复争取时间。本地缓存只适合读多写少、允许短暂不一致的数据，写操作要先清本地缓存再清 Redis，避免长期脏读。

3. 限流、熔断和降级要提前准备好

缓存不可用时，不要让所有流量无限制回源。做法包括：对回源接口做令牌桶限流，超过阈值直接返回降级数据；数据库侧设置慢查询告警和连接池上限；核心接口准备静态兜底页或默认值。演练时可以直接停掉 Redis 观察系统表现，而不是等真实故障时才发现降级开关没生效。

六、落地检查清单

1. 热点 Key 是否有互斥锁或逻辑过期，而不是裸查库。
2. 锁是否有过期时间、随机标识和 Lua 安全释放。
3. 不存在的数据是否有空值短缓存，非法参数是否提前拦截。
4. TTL 是否加了随机抖动，批量刷新是否分批。
5. 是否有缓存命中率、回源 QPS、锁等待时间、数据库慢查询四类监控和告警。
6. 是否做过缓存宕机演练，确认限流和降级确实生效。

七、常见踩坑

只加互斥锁却不设锁过期时间，重建失败会把所有请求堵死；布隆过滤器不做增量更新，新上线的商品长期被误判为不存在；本地缓存不设容量上限，内存被打爆；把库存、订单这类强一致数据也用逻辑过期返回旧值，导致超卖或价格错误。缓存策略没有银弹，关键是按数据一致性要求分级：可容忍短暂旧值的数据用逻辑过期，强一致数据走锁加实时校验，不存在的数据用空值缓存兜底。