title: "Redis 缓存策略：从 Cache Aside 到 Write Behind 的实战选择"
date: "2026-07-10"
tags: ["Redis", "缓存", "架构设计", "性能优化"]
Redis 缓存策略：从 Cache Aside 到 Write Behind 的实战选择
缓存是提升系统性能最直接的手段。但"加个 Redis"只是起点，真正的难点在于选择哪种缓存策略。
五种主流策略
1. Cache Aside（旁路缓存）
最常用的模式。读时先查缓存，miss 则查数据库并回填；写时先更新数据库，再删除缓存。
def get_user(user_id: str) -> dict:
cached = redis.get(f"user:{user_id}")
if cached:
return json.loads(cached)
user = db.query("SELECT * FROM users WHERE id = %s", user_id)
if user:
redis.setex(f"user:{user_id}", 3600, json.dumps(user))
return user
def update_user(user_id: str, data: dict):
db.execute("UPDATE users SET ... WHERE id = %s", user_id, data)
redis.delete(f"user:{user_id}")适用场景：读多写少，对一致性要求不极端。
2. Read Through
应用不直接访问数据库，由缓存层代理读操作。缓存 miss 时自动加载数据。
class CacheThroughProxy:
def __init__(self, cache, loader):
self.cache = cache
self.loader = loader
def get(self, key: str):
value = self.cache.get(key)
if value is None:
value = self.loader(key)
self.cache.setex(key, 3600, value)
return value适用场景：希望缓存逻辑对业务代码透明。
3. Write Through
写操作同时写入缓存和数据库，由缓存层代理写入。
def write_through(key: str, value: str):
pipeline = redis.pipeline()
pipeline.setex(key, 3600, value)
pipeline.execute()
db.execute("INSERT INTO cache_store (key, value) VALUES (%s, %s)", key, value)适用场景：读写比例均衡，需要缓存和数据库强一致。
4. Write Behind（Write Back）
写操作只更新缓存，异步批量同步到数据库。性能最高，但有数据丢失风险。
import asyncio
from collections import defaultdict
write_buffer = defaultdict(dict)
async def write_behind(key: str, value: str):
write_buffer[key] = value
redis.setex(key, 3600, value)
async def flush_to_db():
while True:
await asyncio.sleep(5)
if write_buffer:
batch = dict(write_buffer)
write_buffer.clear()
db.executemany(
"INSERT INTO cache_store (key, value) VALUES (%s, %s) ON CONFLICT UPDATE",
list(batch.items())
)适用场景：写密集场景，可容忍短暂数据丢失（如日志、计数器）。
5. Refresh Ahead
缓存即将过期时，异步刷新数据。避免用户请求触发缓存加载。
def get_with_refresh(key: str, ttl: int = 3600) -> str:
value = redis.get(key)
remaining_ttl = redis.ttl(key)
if remaining_ttl < ttl * 0.2: # 剩余不足 20%
asyncio.create_task(refresh_cache(key))
return value
async def refresh_cache(key: str):
new_value = await load_from_source(key)
redis.setex(key, 3600, new_value)适用场景：热点数据，不希望用户感知缓存 miss。
策略对比
| 策略 | 读延迟 | 写延迟 | 一致性 | 复杂度 | 数据丢失风险 |
|------|--------|--------|--------|--------|-------------|
| Cache Aside | 低 | 低 | 最终一致 | 低 | 无 |
| Read Through | 低 | - | 最终一致 | 中 | 无 |
| Write Through | - | 中 | 强一致 | 中 | 无 |
| Write Behind | - | 极低 | 弱 | 高 | 有 |
| Refresh Ahead | 极低 | - | 最终一致 | 中 | 无 |
常见问题
缓存穿透
查询不存在的数据，每次都打到数据库。
# 方案 1：缓存空值
def get_user_safe(user_id: str):
cached = redis.get(f"user:{user_id}")
if cached == "NULL":
return None
if cached:
return json.loads(cached)
user = db.query("SELECT * FROM users WHERE id = %s", user_id)
if user:
redis.setex(f"user:{user_id}", 3600, json.dumps(user))
else:
redis.setex(f"user:{user_id}", 60, "NULL") # 短 TTL
return user
# 方案 2：布隆过滤器
from pybloom_live import BloomFilter
user_bloom = BloomFilter(capacity=1000000, error_rate=0.01)
def check_exists(user_id: str) -> bool:
if user_id not in user_bloom:
return False
return True缓存雪崩
大量缓存同时过期，请求全部打到数据库。
import random
def set_with_jitter(key: str, value: str, base_ttl: int = 3600):
jitter = random.randint(-300, 300) # ±5 分钟随机
redis.setex(key, base_ttl + jitter, value)缓存击穿
热点 key 过期瞬间，大量并发请求同时查数据库。
from redis import Redis
def get_with_lock(key: str, loader, ttl: int = 3600):
value = redis.get(key)
if value:
return value
lock_key = f"lock:{key}"
if redis.set(lock_key, "1", nx=True, ex=10):
try:
value = loader(key)
redis.setex(key, ttl, value)
finally:
redis.delete(lock_key)
else:
time.sleep(0.1)
return get_with_lock(key, loader, ttl)
return value选型建议
- **通用场景**：Cache Aside，简单可靠
- **读密集 + 热点数据**：Cache Aside + Refresh Ahead
- **写密集 + 可容忍丢失**：Write Behind
- **强一致性要求**：Write Through 或不用缓存
缓存策略没有银弹，关键是根据业务特点选择最合适的方案。
读者评论 2