缓存击穿的本质与危害
Redis缓存击穿是指某个热点Key在缓存中过期的瞬间，大量并发请求同时穿透到数据库，导致数据库瞬时负载飙升甚至宕机。与缓存穿透（查询不存在的Key）和缓存雪崩（大量Key同时过期）不同，缓存击穿针对的是单个或少数几个访问量极高的Key——可能是一条热门商品数据，也可能是一份高频访问的配置项。
数据库运维和MySQL性能调优实践中，缓存击穿是最常见的线上故障诱因之一。一次商品秒杀活动中，某个SKU的缓存Key过期后1秒内涌入5000次数据库查询，直接打满连接池，导致整个服务不可用。这类故障在Redis 7.x环境下有成熟的防护方案。
分布式锁方案：互斥重建缓存
缓存击穿防护的核心思路是：只允许一个请求穿透到数据库重建缓存，其余请求等待缓存重建完成后直接读取缓存。Redis分布式锁是实现互斥重建的经典方案：
# Redis 7.x SET命令支持NX和PX参数
# SET key value NX PX timeout
# NX: 仅当Key不存在时设置成功
# PX: 设置毫秒级过期时间
SET lock:product:10086 1 NX PX 3000在应用层实现互斥重建：
import redis
import json
import time
import threading
class CacheService:
def __init__(self, redis_client, db_client):
self.redis = redis_client
self.db = db_client
self.lock_timeout = 3000 # 锁超时3秒
self.cache_ttl = 3600 # 缓存过期1小时
self.retry_interval = 0.05 # 重试间隔50ms
self.max_retries = 60 # 最多重试60次（3秒）
def get_product(self, product_id: str) -> dict:
cache_key = f"product:{product_id}"
lock_key = f"lock:product:{product_id}"
# 1. 查缓存
data = self.redis.get(cache_key)
if data is not None:
return json.loads(data)
# 2. 缓存未命中，尝试获取分布式锁
lock_acquired = self.redis.set(
lock_key, "1", nx=True, px=self.lock_timeout
)
if lock_acquired:
try:
# 双重检查：获取锁后再次查缓存
data = self.redis.get(cache_key)
if data is not None:
return json.loads(data)
# 查询数据库
result = self.db.query(
"SELECT * FROM products WHERE id = %s", (product_id,)
)
if result:
self.redis.set(
cache_key,
json.dumps(result),
ex=self.cache_ttl
)
else:
# 空值缓存，防止穿透
self.redis.set(cache_key, "NULL", ex=60)
return result
finally:
# 释放锁（使用Lua脚本确保原子性）
self._release_lock(lock_key)
else:
# 未获取锁，等待其他线程重建缓存
return self._wait_for_cache(cache_key)
def _release_lock(self, lock_key: str):
"""Lua脚本释放锁，避免误删其他请求的锁"""
script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
self.redis.eval(script, 1, lock_key, "1")
def _wait_for_cache(self, cache_key: str, timeout: float = 3.0):
"""轮询等待缓存重建完成"""
start = time.time()
while time.time() - start < timeout:
data = self.redis.get(cache_key)
if data is not None:
return json.loads(data) if data != b"NULL" else None
time.sleep(self.retry_interval)
# 超时后降级返回
return None分布式锁方案的关键细节：锁超时时间必须大于数据库查询耗时，否则锁提前释放会导致多个请求同时查库；释放锁必须用Lua脚本保证原子性，避免误删其他请求的锁。
热点Key自动识别与本地缓存预加载
分布式锁解决了并发重建问题，但没有解决缓存过期那一刻的短暂延迟。更彻底的方案是在Redis缓存之上增加一层进程内本地缓存（L1缓存），热点Key在本地缓存中始终可用，永不过期：
import time
import threading
from collections import OrderedDict
class LocalCache:
"""LRU本地缓存，容量有限，命中率要求高的热点Key存储"""
def __init__(self, max_size=1000, default_ttl=600):
self.max_size = max_size
self.default_ttl = default_ttl
self._cache = OrderedDict()
self._lock = threading.Lock()
def get(self, key: str):
with self._lock:
item = self._cache.get(key)
if item is None:
return None
if time.time() > item['expire_at']:
del self._cache[key]
return None
self._cache.move_to_end(key) # LRU更新
return item['value']
def put(self, key: str, value, ttl=None):
with self._lock:
if key in self._cache:
del self._cache[key]
elif len(self._cache) >= self.max_size:
self._cache.popitem(last=False) # 淘汰最旧
self._cache[key] = {
'value': value,
'expire_at': time.time() + (ttl or self.default_ttl),
}
class MultiLevelCache:
"""L1本地缓存 + L2 Redis缓存 两级架构"""
def __init__(self, redis_client, db_client):
self.redis = redis_client
self.db = db_client
self.local_cache = LocalCache(max_size=500, default_ttl=300)
def get(self, key: str):
# L1本地缓存
value = self.local_cache.get(key)
if value is not None:
return value
# L2 Redis缓存
data = self.redis.get(key)
if data is not None:
self.local_cache.put(key, json.loads(data))
return json.loads(data)
# 数据库查询并回填两层缓存
result = self._load_from_db(key)
if result:
self.redis.set(key, json.dumps(result), ex=3600)
self.local_cache.put(key, result)
return result本地缓存的过期时间应该短于Redis缓存，形成一个梯度：本地缓存5分钟，Redis缓存1小时。这样即使Redis缓存过期，本地缓存仍能短暂兜底，为Redis缓存重建争取时间。
Redis 7.x Function实现原子性缓存更新
Redis 7引入了Redis Function功能，替代早期的Lua脚本eval方式。Function在服务端注册并持久化，执行时无需传输脚本内容，性能更好：
-- 注册缓存重建函数
-- redis-cli -x FUNCTION LOAD REPLACE < cache_rebuild.js
#!lua name=cache_utils
local function rebuild_cache(keys, args)
local cache_key = keys[1]
local lock_key = keys[2]
local lock_value = args[1]
local lock_ttl = tonumber(args[2])
local cache_data = args[3]
local cache_ttl = tonumber(args[4])
-- 获取锁
local locked = redis.call('SET', lock_key, lock_value, 'NX', 'PX', lock_ttl)
if not locked then
return 0 -- 未获取锁
end
-- 写入缓存
redis.call('SET', cache_key, cache_data, 'EX', cache_ttl)
-- 释放锁
redis.call('DEL', lock_key)
return 1
end
redis.register_function('rebuild_cache', rebuild_cache)# Python调用Redis Function
import redis
r = redis.Redis(host='10.0.1.50', port=6379, db=0)
def rebuild_with_function(product_id: str, data: dict):
result = r.fcall(
'rebuild_cache',
2, # keys数量
f'product:{product_id}', # cache_key
f'lock:product:{product_id}', # lock_key
'unique_lock_value', # lock_value
'3000', # lock_ttl ms
json.dumps(data), # cache_data
'3600' # cache_ttl seconds
)
return result == 1 # 1=成功，0=锁冲突缓存预热与永不过期策略
对于已知的热点Key，采用"逻辑过期"替代TTL物理过期，从根本上杜绝缓存击穿的发生：
class LogicalExpireCache:
"""逻辑过期：缓存永不过期，数据中包含过期时间戳"""
def __init__(self, redis_client, db_client):
self.redis = redis_client
self.db = db_client
def get(self, key: str):
raw = self.redis.get(key)
if raw is None:
# 缓存完全不存在，首次加载
return self._load_and_cache(key)
data = json.loads(raw)
if data.get('expire_at', 0) > time.time():
# 逻辑未过期，直接返回
return data['payload']
# 逻辑已过期，异步重建（返回旧数据兜底）
threading.Thread(
target=self._async_rebuild, args=(key,), daemon=True
).start()
return data['payload'] # 返回过期数据，不影响请求
def _async_rebuild(self, key: str):
lock_key = f"lock:{key}"
locked = self.redis.set(lock_key, "1", nx=True, px=5000)
if locked:
try:
result = self._load_from_db(key)
self._set_logical_expire(key, result, ttl=3600)
finally:
self.redis.delete(lock_key)
def _set_logical_expire(self, key: str, data: dict, ttl: int):
cache_data = {
'payload': data,
'expire_at': time.time() + ttl,
}
self.redis.set(key, json.dumps(cache_data)) # 不设TTL，永不过期逻辑过期方案的要点：Redis中的Key永不过期（不设TTL），过期时间写在Value的JSON字段中。当检测到逻辑过期时，异步线程重建缓存，当前请求返回旧数据。用户感知到的行为是：过期后的第一次请求拿到的是旧数据（可接受的短暂不一致），后续请求拿到新数据。这个方案牺牲了短暂的数据一致性，换来的是零缓存击穿风险。
监控告警：热点Key实时检测
Redis 7的OBJECT FREQ命令配合LFU淘汰策略，可以实时追踪Key的访问频率：
# 开启LFU淘汰策略（redis.conf）
maxmemory-policy allkeys-lfu
# 查看Key访问频率
redis-cli OBJECT FREQ product:10086
# 批量扫描高频Key
redis-cli --scan --pattern "product:*" | while read key; do
freq=$(redis-cli OBJECT FREQ "$key")
if [ "$freq" -gt 10000 ]; then
echo "热点Key: $key, 访问频率: $freq"
fi
done配合Prometheus + Grafana监控Redis的keyspace_hits、keyspace_misses和instantaneous_ops_per_sec指标，当缓存命中率低于95%或瞬时操作数突增3倍以上时触发告警，运维人员可以提前识别热点Key并配置预热策略。
缓存击穿防护没有银弹，分布式锁、本地缓存、逻辑过期三种方案各有取舍。生产环境推荐组合使用：逻辑过期方案覆盖核心热点Key，分布式锁兜底一般Key，本地缓存作为全局加速层。三层防线协同，才能在高并发场景下保证数据库安全。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/redis7-huan-cun-ji-chuan-yu-re-dian-key-zhi-li-cong-fen-bu/
