Redis缓存策略深度实战

Redis 缓存策略深度实战：穿透、击穿、雪崩全面解析

作者：Crown_22| AI Agent & 自动化工作流开发者 | 技术分享

前言

Redis 是后端开发的标配，但你真的会用缓存吗？

我见过太多项目，Redis 用得飞起，结果上线后各种问题：缓存穿透打爆数据库、缓存击穿导致服务雪崩、数据一致性一塌糊涂。今天我把多年实战经验分享出来，帮你避开这些坑。

一、缓存基础模式

1.1 Cache-Aside（旁路缓存）

最常见的缓存模式，读写都走应用层：

importredisimportjsonclassCacheAside:def__init__(self,redis_client,db):self.redis=redis_client self.db=dbdefget(self,key:str):# 1. 先查缓存cached=self.redis.get(key)ifcached:returnjson.loads(cached)# 2. 缓存未命中，查数据库data=self.db.query(key)ifdata:# 3. 写入缓存self.redis.setex(key,3600,# 过期时间1小时json.dumps(data))returndatadefset(self,key:str,value:dict):# 1. 更新数据库self.db.update(key,value)# 2. 删除缓存（而不是更新）self.redis.delete(key)

关键点：写的时候是删除缓存，不是更新缓存。为什么？

# ❌ 错误：更新缓存defset_wrong(self,key,value):self.db.update(key,value)self.redis.set(key,value)# 并发时可能写入脏数据# ✅ 正确：删除缓存defset_correct(self,key,value):self.db.update(key,value)self.redis.delete(key)# 下次读取时会重新加载

1.2 Read-Through / Write-Through

缓存层代理数据库操作：

classReadThroughCache:def__init__(self,redis_client,db):self.redis=redis_client self.db=dbdefget(self,key:str):cached=self.redis.get(key)ifcached:returnjson.loads(cached)# 缓存自己加载数据（对应用透明）data=self.db.query(key)self.redis.setex(key,3600,json.dumps(data))returndata

1.3 Write-Behind（异步写回）

写操作先写缓存，异步批量写入数据库：

importasynciofromcollectionsimportdefaultdictclassWriteBehindCache:def__init__(self,redis_client,db):self.redis=redis_client self.db=db self.dirty_keys=set()self.lock=asyncio.Lock()asyncdefset(self,key:str,value:dict):# 1. 写入缓存self.redis.setex(key,3600,json.dumps(value))# 2. 标记为脏数据asyncwithself.lock:self.dirty_keys.add(key)# 3. 异步批量写入数据库asyncio.create_task(self._flush_to_db())asyncdef_flush_to_db(self):asyncwithself.lock:ifnotself.dirty_keys:returnkeys_to_flush=list(self.dirty_keys)self.dirty_keys.clear()# 批量写入数据库forkeyinkeys_to_flush:data=self.redis.get(key)ifdata:self.db.update(key,json.loads(data))

二、缓存三大问题

2.1 缓存穿透

问题：查询不存在的数据，每次都穿透缓存打到数据库。

# ❌ 恶意请求不存在的IDGET/user/99999999# 数据库中不存在

解决方案1：缓存空值

defget_with_null_cache(self,key:str):cached=self.redis.get(key)ifcached=="NULL":returnNone# 空值缓存ifcached:returnjson.loads(cached)data=self.db.query(key)ifdataisNone:# 缓存空值，较短过期时间self.redis.setex(key,300,"NULL")# 5分钟else:self.redis.setex(key,3600,json.dumps(data))returndata

解决方案2：布隆过滤器

frompybloom_liveimportBloomFilterclassBloomFilterCache:def__init__(self,redis_client,db