缓存失效的三种典型模式
数据库运维中，Redis缓存层是保护后端数据库的第一道防线。当缓存层大面积失效时，请求直接打到数据库，轻则响应变慢，重则数据库连接池耗尽引发级联故障。缓存失效分为三种模式：缓存穿透——查询不存在的数据，缓存和数据库都没有，每次请求都穿透到数据库；缓存击穿——热点Key过期瞬间，大量并发请求同时涌向数据库；缓存雪崩——大量Key同时过期或Redis节点宕机，数据库瞬间承受巨大压力。三种模式的防御策略各有侧重，需要针对性设计。
缓存穿透：布隆过滤器与空值缓存
穿透场景的典型例子：电商系统中用户查询一个不存在的商品ID。缓存中没有，数据库中也没有，但每次查询都会穿透到数据库层。攻击者可以利用这一点构造大量无效查询，耗尽数据库连接。
防御方案一——布隆过滤器。在请求到达Redis之前，先用布隆过滤器判断数据是否可能存在：
import redis
import mmh3
class BloomFilter:
def __init__(self, redis_client, key, capacity=1000000, error_rate=0.001):
self.redis = redis_client
self.key = key
# 计算所需位数组和哈希函数数量
self.size = int(-capacity * (math.log(error_rate) / (math.log(2) ** 2)))
self.hash_count = int((self.size / capacity) * math.log(2))
def _get_offsets(self, value):
offsets = []
v = str(value)
for i in range(self.hash_count):
hash_val = mmh3.hash(v, i) % self.size
offsets.append(hash_val)
return offsets
def add(self, value):
pipe = self.redis.pipeline()
for offset in self._get_offsets(value):
pipe.setbit(self.key, offset, 1)
pipe.execute()
def might_contain(self, value):
pipe = self.redis.pipeline()
for offset in self._get_offsets(value):
pipe.getbit(self.key, offset)
results = pipe.execute()
return all(results)
# 初始化：将所有有效ID加载到布隆过滤器
bloom = BloomFilter(redis_client, "product_ids_bloom")
for product_id in get_all_product_ids():
bloom.add(product_id)
# 查询前先过滤
def get_product(product_id):
if not bloom.might_contain(product_id):
return None # 确定不存在，直接返回
# 可能存在，走正常缓存查询流程
return get_product_from_cache_or_db(product_id)布隆过滤器有误判率——不存在的数据可能被判为存在（false positive），但存在的数据不会被遗漏（no false negative）。误判率可通过调整capacity和error_rate参数控制。
防御方案二——空值缓存。对于查询到数据库也不存在的数据，在Redis中写入空值并设置较短TTL：
def get_product_from_cache_or_db(product_id):
# 1. 查Redis缓存
cached = redis_client.get(f"product:{product_id}")
if cached is not None:
if cached == b"NULL":
return None # 空值缓存命中
return json.loads(cached)
# 2. 查数据库
product = db.query("SELECT * FROM products WHERE id = %s", product_id)
if product is None:
# 写入空值缓存，TTL 5分钟
redis_client.setex(f"product:{product_id}", 300, "NULL")
return None
# 3. 正常缓存，TTL 1小时
redis_client.setex(f"product:{product_id}", 3600, json.dumps(product))
return product缓存击穿：互斥锁与逻辑过期
热点Key过期时，几百个并发线程同时发现缓存失效，同时去数据库加载数据。解决方案是在缓存重建时加互斥锁，只允许一个线程去数据库查询，其余线程等待：
import redis
import time
import threading
class CacheWithMutex:
def __init__(self, redis_client):
self.redis = redis_client
self._local_locks = {}
self._lock = threading.Lock()
def get(self, key, db_loader, ttl=3600):
value = self.redis.get(key)
if value is not None:
return json.loads(value)
# 获取进程内锁，防止同进程多线程并发重建
with self._lock:
if key not in self._local_locks:
self._local_locks[key] = threading.Lock()
lock = self._local_locks[key]
with lock:
# Double check
value = self.redis.get(key)
if value is not None:
return json.loads(value)
# 分布式锁 - 防止多节点并发重建
lock_key = f"lock:{key}"
lock_acquired = self.redis.set(
lock_key, "1", nx=True, ex=10
)
if lock_acquired:
try:
data = db_loader()
self.redis.setex(key, ttl, json.dumps(data))
return data
finally:
self.redis.delete(lock_key)
else:
# 等待其他节点重建完成
for _ in range(50):
time.sleep(0.1)
value = self.redis.get(key)
if value is not None:
return json.loads(value)
raise Exception("Cache rebuild timeout")逻辑过期是另一种方案——缓存永不过期（不设TTL），在值中存储逻辑过期时间，后台线程异步刷新：
def get_with_logical_expire(key, db_loader, expire_seconds=3600):
data = redis_client.get(key)
if data is None:
# 首次加载
value = db_loader()
cache_data = {
"value": value,
"expire_at": time.time() + expire_seconds
}
redis_client.set(key, json.dumps(cache_data))
return value
cache_obj = json.loads(data)
if time.time() < cache_obj["expire_at"]:
return cache_obj["value"] # 逻辑未过期
# 逻辑过期，触发异步刷新，同时返回旧数据
threading.Thread(
target=_async_rebuild,
args=(key, db_loader, expire_seconds),
daemon=True
).start()
return cache_obj["value"] # 返回过期但可用的数据
def _async_rebuild(key, db_loader, expire_seconds):
lock_key = f"lock:{key}"
if redis_client.set(lock_key, "1", nx=True, ex=10):
try:
value = db_loader()
cache_data = {
"value": value,
"expire_at": time.time() + expire_seconds
}
redis_client.set(key, json.dumps(cache_data))
finally:
redis_client.delete(lock_key)缓存雪崩：过期时间打散与多级缓存
雪崩的根源是大量Key同时过期。预防措施是在TTL基础上添加随机偏移量：
import random
def set_with_jitter(key, value, base_ttl=3600, jitter_range=300):
jitter = random.randint(0, jitter_range)
actual_ttl = base_ttl + jitter
redis_client.setex(key, actual_ttl, json.dumps(value))批量设置缓存时务必使用jitter：
# 错误写法 - 所有Key同一时刻过期
for product in products:
redis_client.setex(f"product:{product.id}", 3600, json.dumps(product))
# 正确写法 - 过期时间分散在50-70分钟
for product in products:
ttl = random.randint(3000, 4200)
redis_client.setex(f"product:{product.id}", ttl, json.dumps(product))Redis节点宕机场景下，多级缓存提供兜底保障：
class MultiLevelCache:
def __init__(self):
self.redis = redis_client
self.local_cache = {} # 进程内缓存
self.local_ttl = 60 # 本地缓存60秒
def get(self, key, db_loader):
# L1: 进程内缓存
local_data = self.local_cache.get(key)
if local_data and time.time() < local_data["expire_at"]:
return local_data["value"]
# L2: Redis缓存
try:
data = self.redis.get(key)
if data is not None:
value = json.loads(data)
self._set_local(key, value)
return value
except redis.ConnectionError:
pass # Redis不可用，降级到数据库
# L3: 数据库
value = db_loader()
try:
self.redis.setex(key, 3600, json.dumps(value))
except redis.ConnectionError:
pass
self._set_local(key, value)
return value
def _set_local(self, key, value):
self.local_cache[key] = {
"value": value,
"expire_at": time.time() + self.local_ttl
}监控指标与告警阈值
缓存防御体系需要配套监控才能闭环。关键指标：缓存命中率（hit_rate）低于85%需要排查穿透问题；Redis连接数接近maxclients时检查是否穿透流量激增；数据库连接池使用率超过80%且缓存命中率同步下降，意味着雪崩正在发生。Prometheus告警规则：
- alert: CacheHitRateLow
expr: redis_keyspace_hits_total / (redis_keyspace_hits_total + redis_keyspace_misses_total) < 0.85
for: 5m
labels:
severity: warning
annotations:
summary: "Redis缓存命中率低于85%"把布隆过滤器、互斥锁、TTL打散、多级缓存四层防御组合使用，构建完整的缓存安全网。任何单一方案都无法覆盖所有失效场景，工程化的缓存防御必须是纵深体系。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/redis-huan-cun-chuan-tou-yu-xue-beng-de-gong-cheng-hua-fang/
