Python 缓存策略：Redis 缓存失效、缓存击穿、缓存雪崩解决方案
Introduction
缓存是提升 Web 应用性能的杀手锏，但用不好反而会引入新问题。Redis 缓存有三种经典坑：缓存失效（缓存过期后大量请求同时打穿数据库）、缓存击穿（热点 key 过期瞬间大量并发请求直达 DB）、缓存雪崩（大量 key 集中过期导致数据库被打爆）。本文用 Python + Redis 演示三种场景的成因和应对方案。
---
1. 缓存失效（Cache Miss Storm）
场景
高并发下，某个热点数据缓存过期后，大量请求同时发现缓存不存在，全部去查数据库，导致数据库负载瞬间飙升。
演示
import redis, time, random, threading
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# ---------- 问题演示：无锁 ----------
def get_user_no_lock(user_id):
"""没有并发保护的查库——大量并发时会产生缓存失效风暴"""
cache_key = f"user:{user_id}"
cached = r.get(cache_key)
if cached:
return f"缓存命中: {cached}"
# 模拟数据库查询（100ms）
time.sleep(0.1)
data = f"user_data_{user_id}_{random.randint(1000,9999)}"
# 写入缓存
r.setex(cache_key, ttl=1, value=data) # TTL=1秒，故意设短模拟过期
return f"数据库查询: {data}"
# ---------- 解决方案：分布式锁 ----------
from contextlib import contextmanager
_lock_redis = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
@contextmanager
def distributed_lock(key, timeout=5):
"""Redis SETNX 实现简单分布式锁"""
lock_key = f"lock:{key}"
acquired = _lock_redis.set(lock_key, "1", nx=True, ex=timeout)
if not acquired:
# 未拿到锁，等待锁释放后重试（最多3次）
for _ in range(3):
time.sleep(0.05)
if _lock_redis.get(lock_key) is None:
break
else:
raise RuntimeError(f"获取锁失败: {key}")
try:
yield
finally:
_lock_redis.delete(lock_key)
def get_user_with_lock(user_id):
"""加锁后查库，同一时刻只有一个请求能查数据库"""
cache_key = f"user:{user_id}"
cached = r.get(cache_key)
if cached:
return f"缓存命中: {cached}"
try:
with distributed_lock(cache_key):
# 双重检查（拿到锁后再次确认缓存）
cached = r.get(cache_key)
if cached:
return f"双重检查缓存命中: {cached}"
time.sleep(0.1) # 模拟 DB 查询
data = f"user_data_{user_id}_{random.randint(1000,9999)}"
r.setex(cache_key, ttl=60, value=data)
return f"数据库查询并写入缓存: {data}"
except RuntimeError as e:
# 锁获取失败，降级为直接查库
return f"降级查库: {e}"
运行效果
无锁版本：TTL=1秒，过期后多个并发请求同时 miss，全部去查数据库。加锁版本：同一时刻只有一个请求能查库，其他请求等锁释放后直接读缓存。
---
2. 缓存击穿（Hot Key Expiration）
场景
某个 key 是热点数据（比如微博热搜），过期瞬间有 10 万并发请求进来，全部打到数据库，数据库直接挂掉。
解决方案对比
| 方案 | 原理 | 优点 | 缺点 |
|------|------|------|------|
| 分布式锁 | 只允许一个请求查库 | 彻底解决问题 | 有锁开销 |
| 永不过期 + 异步重建 | value 存结构体（含数据+时间戳），异步更新 | 无锁，无失效期 | 实现复杂，空间占用大 |
| 逻辑过期 | 热点 key 过期后返回旧数据，同时异步更新 | 体验好 | 旧数据窗口期 |
演示：逻辑过期方案
import json, time, random
from threading import Thread
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# ---------- 预热：写入带逻辑过期时间的数据 ----------
def init_hot_key():
data = {
"user_id": 42,
"username": "alice",
"votes": 9527,
"logic_expire_at": time.time() + 5 # 逻辑过期时间：5秒后
}
r.set("hot:topic:1", json.dumps(data))
print(f"预热完成，当前 votes={data['votes']}，逻辑过期时间={data['logic_expire_at']}")
# ---------- 异步重建缓存 ----------
def async_rebuild():
time.sleep(0.2) # 模拟 DB 查询耗时
new_data = {
"user_id": 42,
"username": "alice",
"votes": 9999, # 新数据
"logic_expire_at": time.time() + 5
}
r.set("hot:topic:1", json.dumps(new_data))
print(f"[异步重建] votes 更新为 {new_data['votes']}")
# ---------- 读数据（逻辑过期） ----------
def get_hot_data():
raw = r.get("hot:topic:1")
if not raw:
return None
data = json.loads(raw)
# 逻辑过期：缓存有效但时间戳已过
if time.time() > data["logic_expire_at"]:
# 启动异步重建（不等完成直接返回旧数据）
Thread(target=async_rebuild, daemon=True).start()
return data # 仍返回旧数据，体验上只是"稍旧"
return data
# ---------- 演示 ----------
init_hot_key()
print("\n=== 模拟 3 次请求（5秒内）===")
for i in range(3):
result = get_hot_data()
print(f"请求{i+1} 返回: votes={result['votes']}")
time.sleep(2)
print("\n=== 5秒后再次请求（逻辑已过期）===")
time.sleep(2)
result = get_hot_data()
print(f"请求4 返回: votes={result['votes']}")
运行效果
前 3 次请求（5秒内）直接读缓存，返回 votes=9527。第 4 次请求时逻辑已过期，触发异步重建，返回 votes=9527（旧数据），后台同时更新缓存，更新后 votes=9999。
---
3. 缓存雪崩（Cache Avalanche）
场景
大量 key 同时过期（TTL 设成同一值），导致某一时刻大量请求同时 miss，全部打向数据库。
解决方案
方案 A：TTL 随机化，不让大量 key 同时过期：
import random
def set_with_jitter(key, value, base_ttl=3600):
"""在 base_ttl 基础上加随机抖动（0 ~ 300秒），防止雪崩"""
ttl = base_ttl + random.randint(0, 300)
r.setex(key, ttl=ttl, value=value)
print(f"写入 {key}，TTL={ttl}秒")
# 批量写入时加随机抖动
for i in range(100):
set_with_jitter(f"product:{i}", f"product_data_{i}", base_ttl=3600)
方案 B：二级缓存，本地内存做备份：
from functools import wraps
import redis, json, time, threading
_local_cache = {} # 本地缓存 dict（生产环境用 LRU Cache）
def get_with_local_cache(redis_conn, key, ttl=60):
"""先查本地内存，再查 Redis，最后查 DB"""
# 1. 本地缓存
if key in _local_cache:
val, expire_at = _local_cache[key]
if time.time() < expire_at:
return f"[本地缓存] {val}"
# 2. Redis
val = redis_conn.get(key)
if val:
_local_cache[key] = (val, time.time() + 10) # 本地缓存10秒
return f"[Redis缓存] {val}"
# 3. 数据库（模拟）
val = f"db_data_for_{key}"
redis_conn.setex(key, ttl=ttl, value=val)
return f"[数据库查询] {val}"
方案 C：服务熔级降级，数据库撑不住时直接返回默认值：
from typing import Optional
def get_or_default(user_id: int, default: str = "数据加载中") -> str:
cache_key = f"user:{user_id}"
try:
val = r.get(cache_key)
if val:
return val
# 数据库查询（省略）
val = f"user_data_{user_id}"
r.setex(cache_key, ttl=3600, value=val)
return val
except redis.exceptions.ConnectionError:
# Redis 连不上，降级返回默认值
print("Redis 连接失败，降级返回默认值")
return default
except Exception as e:
print(f"查询异常: {e}，降级返回默认值")
return default
---
常见问题
Q1: 锁的 key 和数据的 key 能一样吗？ 不能。锁的 key 是 lock:user:42，数据的 key 是 user:42，两者分开。如果混了，删除锁时会误删数据。
Q2: 分布式锁用什么 Redis 命令？ 用 SET key value NX EX timeout，这是原子操作。不要用 SETNX + EXPIRE 两步（两步之间进程崩溃会导致锁永久不释放）。
Q3: 本地缓存和 Redis 缓存怎么选？ 小数据用本地缓存（进程内，无网络开销），大数据用 Redis。本地缓存注意设置 LRU（最近最少使用）上限，防止内存溢出。
Q4: 缓存雪崩和缓存击穿有什么区别？ 雪崩是大量 key 同时过期造成批量 DB 压力；击穿是单个热点 key 过期造成瞬时 DB 压力。解决方案不同：雪崩靠 TTL 随机化，击穿靠锁或逻辑过期。
Q5: Redis 内存满了怎么办？ 配 maxmemory-policy（淘汰策略）。推荐 allkeys-lru（优先淘汰最近最少使用的 key）；对实时性要求高的场景用 volatile-lru（只淘汰设了过期时间的 key）。
---
延伸阅读
- <a href="/category/12">Python Redis 快速入门：安装、基础命令、Python 操作实战</a> — Redis 基础操作
- <a href="/category/8">Python 标准库专题</a> — 更多 Python 性能相关教程
- <a href="/category/12">数据库连接池专题</a> — DB 连接优化实战
---
作者：小马 | 绍大技术网 shaoda.net