企业官网建设流程全解析

缓存穿透、击穿、雪崩：三个经典问题与完整解决方案

这三个问题是 Redis 面试必考，也是线上事故高发区。
名字相似但成因和解决方式完全不同，混着说一定是没真懂。

本篇把三者的成因、区别、解决方案讲透，
每种方案都给可直接用的代码。

一、先分清三个概念

快速区分：

• 查的是不存在的 id（比如 id = -1）→ 穿透
• 查的是存在的热点，只是恰好过期 → 击穿
• 一大批key 一起没了 → 雪崩

二、缓存穿透

2.1 成因

defget_user(user_id):data=redis.get(f"user:{user_id}")ifdataisNone:data=db.query("SELECT * FROM users WHERE id = %s",user_id)ifdata:redis.setex(f"user:{user_id}",3600,json.dumps(data))returndata# ← 查不到时，不写缓存，直接返回 Nonereturnjson.loads(data)

问题在最后一步：DB 查不到时不写缓存。
于是每次请求同一个不存在的 id，都会打到 DB。

正常业务中这类请求很少，但如果是恶意攻击
（用脚本遍历不存在的 id），DB 会被打垮。

2.2 解决方案一：缓存空值

defget_user(user_id):key=f"user:{user_id}"data=redis.get(key)ifdataisnotNone:returnNoneifdata==b"__NULL__"elsejson.loads(data)data=db.query("SELECT * FROM users WHERE id = %s",user_id)ifdata:redis.setex(key,3600,json.dumps(data))else:# ✅ 空值也缓存，但 TTL 要短（比如 60 秒）redis.setex(key,60,"__NULL__")returndata

要点：

• 用一个特殊标记（__NULL__）表示"确实不存在"
• TTL 要短（60 秒 vs 正常的 3600 秒），
  否则万一后来真插入了这条数据，会长时间读不到

代价：会占用一些内存存空值。如果攻击者用海量随机 id，
仍会产生大量空 key（所以还要配合下面的方案）。

2.3 解决方案二：布隆过滤器

原理：把所有存在的 key预先放进一个位数组，
查询时先过一遍——布隆过滤器说"不存在"，那就一定不存在，
直接返回，不用查 DB。

# pip install pybloom-live 或用 Redis 的 BF.* 模块frompybloom_liveimportScalableBloomFilter bf=ScalableBloomFilter(initial_capacity=100000,error_rate=0.001)# 初始化：把所有有效 id 加进去foruidindb.query_all_ids():bf.add(uid)defget_user(user_id):ifuser_idnotinbf:# ✅ 一定不存在，直接返回returnNone# 后续照常走缓存逻辑...

用 Redis 原生布隆过滤器（需要 RedisBloom 模块）：

BF.RESERVE user_filter0.0011000000# 误判率 0.1%，容量 100 万BF.ADD user_filter1001BF.EXISTS user_filter1001# 1 = 可能存在BF.EXISTS user_filter-1# 0 = 一定不存在

关键特性：

• ✅说"不存在"就一定不存在（不会漏判）
• ⚠️说"存在"可能其实不存在（有误判率，通常设 0.1%~1%）
• ❌不支持删除（标准布隆过滤器）；要删除得用 Counting Bloom Filter

适用场景：id 集合相对稳定（不会频繁新增）。
如果要频繁新增，需要维护过滤器和 DB 的一致性。

2.4 解决方案三：参数校验 + 限流

defget_user(user_id):# 1. 基础校验：id 必须为正整数且在合理范围ifnotisinstance(user_id,int)oruser_id<=0oruser_id>10**9:returnNone# 2. 对同一 IP 的请求限流ifnotrate_limiter.allow(request.ip):raiseTooManyRequests()...

这是第一道防线，成本最低。
比如 id 不可能是负数、不可能超过 10 亿，直接在接口层拦掉。

2.5 三种方案对比

生产建议：参数校验 + 缓存空值（成本最低，覆盖大多数场景）。
数据量特别大且 id 稳定时再加布隆过滤器。

三、缓存击穿

3.1 成因

某个极热点的 key（比如首页推荐、爆款商品）在某一刻过期，
此时成千上万个请求同时发现缓存没了，
全部涌向 DB 去重建缓存。

区别穿透的关键：这个数据是存在的，只是缓存恰好失效。

3.2 解决方案一：互斥锁（最常用）

只允许一个线程去 DB 查并重建缓存，其他线程等待。

importuuid,timeimportredis r=redis.Redis()defget_with_mutex(key,ttl=3600,lock_ttl=3):data=r.get(key)ifdata:returndata lock_key=f"lock:{key}"token=str(uuid.uuid4())# 尝试获取锁（NX + EX 是原子操作）ifr.set(lock_key,token,nx=True,ex=lock_ttl):try:data=db_query(key)# 只有拿到锁的去查 DBifdata:r.setex(key,ttl,data)else:r.setex(key,60,"__NULL__")# 顺带防穿透returndatafinally:# ⚠️ 必须用 Lua 保证"判断 value 再删"的原子性r.eval(""" if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) end return 0 """,1,lock_key,token)else:# 没拿到锁：短暂等待后重试time.sleep(0.05)returnget_with_mutex(key,ttl,lock_ttl)

🔑三个关键点：

1. SET lock_key token NX EX 3——NX（不存在才设）+ EX（过期时间）必须一起用，
   否则进程崩溃锁永不释放，死锁
2. 释放锁必须用 Lua 脚本比对 token 再删。
   否则可能删掉别人的锁（自己的锁超时了，别人已经拿到锁，
   你一删把别人的删了）
3. lock_ttl要大于 DB 查询耗时，但不要太大

3.3 解决方案二：逻辑过期（永不过期）

不给 key 设物理 TTL，而是把过期时间存进 value 里。

importjson,timedefset_logical(key,value,logical_ttl=3600):payload={"value":value,"expire_at":time.time()+logical_ttl,}r.set(key,json.dumps(payload))# 注意：不设 EXdefget_logical(key):raw=r.get(key)ifrawisNone:returnNone# 真的没有（走正常流程）payload=json.loads(raw)iftime.time()<payload["expire_at"]:returnpayload["value"]# 没过期，直接返回# 逻辑上过期了：返回旧值，同时异步重建ifr.set(f"lock:{key}",1,nx=True,ex=3):threading.Thread(target=rebuild,args=(key,)).start()returnpayload["value"]# ✅ 先返回旧数据，不阻塞

优点：

• 永远不会有"大量请求同时等待"的情况
• 请求不会阻塞，体验好

缺点：

• 会短暂返回过期数据（最终一致）
• 实现复杂，要处理异步重建的并发

适用：对一致性要求不高、但并发极高的场景（比如商品详情、排行榜）。

3.4 两种方案怎么选

四、缓存雪崩

4.1 成因

大批 key 在同一时刻集体过期，
或者Redis 实例直接挂了，
导致所有请求瞬间打到 DB。

常见触发场景：

• 上线时批量预热缓存，全都设了相同的 TTL
• 定时任务在整点刷新缓存
• Redis 集群宕机

4.2 解决方案一：TTL 加随机值

importrandomdefset_with_jitter(key,value,base_ttl=3600):# 基础 TTL + 随机 0~300 秒的抖动ttl=base_ttl+random.randint(0,300)r.setex(key,ttl,value)

就这么简单，但极其有效。
原本 10 万个 key 都在 3600 秒后同时失效，
加上随机抖动后，它们会分散在 3600~3900 秒之间陆续过期，
DB 的压力从"一瞬间的尖峰"变成"平滑的小坡"。

4.3 解决方案二：多级缓存

请求 → 本地缓存(Caffeine/Guava) → Redis → DB

即使 Redis 全挂，本地缓存还能挡一部分。
代价是一致性更难保证（本地缓存无法主动失效，
只能靠短 TTL 或消息通知）。

4.4 解决方案三：Redis 高可用

这是根本解法：

• 主从 + 哨兵：主库挂了自动切从库
• Redis Cluster：分片 + 高可用
• 限流降级：Redis 挂了就限流，保住 DB

# 降级：Redis 不可用时直接返回兜底数据，不去打 DBdefget_with_degrade(key):try:returnr.get(key)exceptredis.ConnectionError:log.error("Redis 不可用，走降级")returnget_fallback(key)# 返回静态兜底数据或空

4.5 解决方案四：缓存预热

系统启动/大促前，主动把热点数据加载到缓存，
并错开 TTL。

defwarmup():foriteminhot_items:set_with_jitter(f"item:{item.id}",item.to_json())

五、一个完整的缓存封装

把上面所有方案组合起来：

importjson,time,random,uuid,threadingimportredis r=redis.Redis()classCache:NULL="__NULL__"def__init__(self,base_ttl=3600,jitter=300,null_ttl=60,lock_ttl=3):self.base_ttl,self.jitter=base_ttl,jitter self.null_ttl,self.lock_ttl=null_ttl,lock_ttldef_ttl(self):returnself.base_ttl+random.randint(0,self.jitter)# 防雪崩defget(self,key,loader):""" loader: 缓存未命中时的 DB 查询函数，返回 None 表示不存在 """raw=r.get(key)# 命中（含空值标记）ifrawisnotNone:returnNoneifraw.decode()==self.NULLelsejson.loads(raw)# 未命中 → 互斥锁重建（防击穿）lock_key=f"lock:{key}"token=str(uuid.uuid4())ifr.set(lock_key,token,nx=True,ex=self.lock_ttl):try:data=loader()# 空值也缓存（防穿透）ifdataisNone:r.setex(key,self.null_ttl,self.NULL)else:r.setex(key,self._ttl(),json.dumps(data))returndatafinally:r.eval(""" if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) end return 0 """,1,lock_key,token)# 没拿到锁 → 等一下重试time.sleep(0.05)returnself.get(key,loader)# 使用cache=Cache()user=cache.get(f"user:{uid}",lambda:db_find_user(uid))

这一个get方法同时解决了：

• 穿透（空值缓存）
• 击穿（互斥锁）
• 雪崩（TTL 随机抖动）

六、三个问题的排查方法

# 监控缓存命中率INFO stats# keyspace_hits: 1000000# keyspace_misses: 50000# 命中率 = hits / (hits + misses) = 95%

命中率突然下降是最直接的信号。
正常业务应该在 90% 以上。

七、小结

三个必须记住的细节：

1. 空值缓存的 TTL 要短（60 秒），否则新增数据读不到
2. 释放锁必须用 Lua 比对 token，否则可能删掉别人的锁
3. TTL 随机抖动是成本最低、收益最高的雪崩防护

下一篇讲分布式锁——本篇的互斥锁只是一个简化版，
真正的分布式锁还要考虑可重入、自动续期、Redlock 等一堆问题。