Redis 缓存三大问题：穿透、击穿、雪崩
缓存明明加了，数据库还是被打挂。三个名字都在说这件事，穿透、击穿、雪崩，但它们的成因、流量形状和防御完全不一样，混着治就是白花钱。分界点只有一个：缓存是被什么请求打穿的。
三个问题的分界
穿透，查的是缓存里根本不会有的数据。通常是恶意请求拿不存在的 key 来回刷，每次缓存都未命中，请求全部漏到数据库。流量形状是「查不存在的东西」。
击穿，某个热点 key 突然过期。大量并发正围着它转，缓存一空，请求同时砸向数据库去重建。流量形状是「一个热点塌了」。
雪崩，大批 key 在同一时刻过期，或者 Redis 整个挂掉。缓存的命中率一夜归零，数据库面对全量流量。流量形状是「缓存整体失效」。
三个箭头殊途同归，数据库接不住。但防御得一个一个来。
穿透：让「不存在」也有答案
最便宜的防御是缓存空值。查不到也把「空」写进缓存，设个短过期，后面的请求拿到空就返回，不再打到数据库。代价是存储被没用的空值占用，以及业务真把数据创建出来时要记得删掉空值，否则新数据被空值挡着。
量大上布隆过滤器。它是一种概率型集合，问「元素在不在」，它说「肯定不在」就一定不在，说「大概在」有误判率。把所有合法 key 预先灌进去，请求先过它，说不在的直接拒。代价是误判率随元素增多上升，而且删除困难，元素下线时过滤器里的痕迹清不掉。
击穿：重建只放一个人进去
防御的核心是互斥。热点 key 过期瞬间，只放一个请求去数据库重建，其他请求等着，重建完成读缓存。单机一个锁就够，分布式环境换分布式锁或者用 Redis 的 SETNX 做同样的事。
另一条路是逻辑过期。缓存的 value 里带一个过期时间字段，key 本身不设 TTL。读的时候发现逻辑过期了，返回旧数据，同时异步去重建。体验好，代价是有一段时间读到的是旧数据，以及重建期间旧数据一直被返回。
两种各有适用。交易类数据要准确，用互斥，宁可等。展示类数据要快，用逻辑过期，容忍片刻旧值。
雪崩：把失效摊开，把流量兜住
第一道防线是过期时间加随机抖动。基础 TTL 后面拼一个随机量，把一批 key 的失效摊到一段时间里，别约好了一起走。
第二道是多级缓存。本地缓存（Caffeine 之类）扛在 Redis 前面，Redis 挂了还有本地这层。代价是一致性变难，本地缓存要设短过期或者监听失效。
第三道是兜底的限流和降级。缓存全没了的时候，数据库的保护靠限流，砍掉一部分请求，保住核心链路。这一步和线程池的拒绝策略是同一个思想，防线消失之前先想好谁兜底。原则就一条：缓存挡不住的请求，要在到数据库之前被拦下。
亲手看两个现场
下面用一段 Python 复现穿透和击穿，并对比防御前后的数据库压力。直接运行 python cache_demo.py，看 DB 被打的次数怎么从 20 变成 1。
# 缓存三问题的现场与防御。直接运行：python cache_demo.py
import threading
import time
class FakeDB:
def __init__(self):
self.calls = 0
self._lock = threading.Lock()
def query(self, key):
with self._lock:
self.calls += 1
time.sleep(0.02) # 模拟一次慢查询
return None if key.startswith("ghost") else f"data-{key}"
db = FakeDB()
cache = {}
cache_lock = threading.Lock() # 保护缓存字典
rebuild_lock = threading.Lock() # 重建互斥
def get_raw(key):
"""裸奔版：缓存没有就打数据库。"""
with cache_lock:
hit = cache.get(key, "MISS")
if hit != "MISS":
return hit
val = db.query(key)
with cache_lock:
cache[key] = val
return val
def get_guarded(key):
"""防御版：空值也缓存，重建只放一个线程。"""
with cache_lock:
hit = cache.get(key, "MISS")
if hit != "MISS":
return hit
with rebuild_lock:
with cache_lock:
hit = cache.get(key, "MISS") # 双重检查，别人可能刚重建完
if hit != "MISS":
return hit
val = db.query(key)
with cache_lock:
cache[key] = val if val is not None else "__EMPTY__"
return val
def burst(fn, key, n=20):
"""20 个线程同时打一个 key。"""
ts = [threading.Thread(target=fn, args=(key,)) for _ in range(n)]
for t in ts:
t.start()
for t in ts:
t.join()
if __name__ == "__main__":
burst(get_raw, "ghost-1")
print(f"穿透裸奔：DB 被打 {db.calls} 次")
cache.clear()
db.calls = 0
burst(get_guarded, "ghost-2")
print(f"穿透加空值缓存：DB 被打 {db.calls} 次")
cache.clear()
db.calls = 0
burst(get_raw, "hot-1")
print(f"击穿裸奔：DB 被打 {db.calls} 次")
cache.clear()
db.calls = 0
burst(get_guarded, "hot-2")
print(f"击穿加互斥：DB 被打 {db.calls} 次")
print("结论：防御就是让到 DB 的请求变成一。")
适用边界
这一篇管单机视角的三问题与防御。Redis 集群的高可用方案、持久化丢了缓存怎么办、以及布隆过滤器的实现选型，是更大的题。分布式锁的实现细节争议多（Redlock 那场争论），本文的互斥用最朴素的锁演示，生产选型另说。记住一句话：三个问题的分界在「谁打穿了缓存」，认错病因，防御就是白加。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/D_ENG_F/article/details/168248933
