1. 项目背景

业务场景

食光集市"川味狂欢节"活动上线当天中午 12:00，流量峰值时刻——大屏监控上 MySQL 的 CPU 从 15% 飙到 98%。DBA 紧急介入，发现一个 SQL 的执行次数高达每秒 4000 次：

SELECT*FROMdishesWHEREstore_id='ST001'ANDis_available=1;

这是每家门店的菜单查询——北京朝阳店（ST001）是最大的门店，每天承载 60% 的流量。4000 QPS 打到数据库，其中 3800 次都是完全重复的查询——菜单每 5 分钟才更新一次，但代码里每次首页刷新都要查一次数据库。

这不仅是性能的问题——当数据库被打满后，支付回调、库存扣减这些真正重要的事务也被排队阻塞。一顿操作下来，MySQL 连接池耗尽，整个订单系统进入"假死"状态 12 分钟。

与此同时，运营想给川味狂欢节增加一个"限时秒杀"功能——每个人限购一份，需要做分布式并发控制。开发团队开始思考：用数据库锁？不行，太慢。用 Python 线程锁？不行，多台服务器之间互相看不见。

答案指向 Redis——一套工具同时解决两个问题：缓存加速 + 分布式锁。

痛点

没有缓存的数据库直连架构，在高并发下会出现：

1. 数据库被打爆：重复查询占满连接池，真正的写操作（下单、支付）无法获取连接，用户看到超时 + 扣款成功但订单未生成。
2. 热点数据雪崩：缓存过期那一刻，大量请求同时穿透到数据库——瞬间流量打到后端，比没缓存时更致命。
3. 缓存与数据库不一致：更新了数据库但忘了删缓存，用户看到的价格是旧价格，下单后实际扣款是新价格——"标价 28 元，付款 32 元"的投诉。
4. 多实例无锁可加：Python 的threading.Lock只在同一个进程内有效。部署了 4 个服务实例，需要一把让 4 个进程都看到的分布式锁。

2. 项目设计

场景：MySQL CPU 98% 的红色告警大屏下，DBA 小马冲进会议室：“ST001 的菜单查询一秒 4000 次，你们是不是又在 for 循环里查数据库了？”

小胖：“不是不是，就是正常的首页菜单查询——只是 ST001 流量太大了。我觉得加个缓存就行了——Redis 里放一份菜单，请求来了先从 Redis 查，查不到再查 MySQL。这不就简单了？”

小白：“小胖你只说了最简单的情况。缓存世界里有很多’陷阱’——缓存穿透（查不存在的数据）、缓存击穿（热点 key 过期瞬间大量并发查 DB）、缓存雪崩（大量 key 同时过期）。这些不是’加个 Redis 就完事’的问题。而且——缓存和数据库的一致性怎么保证？先删缓存还是先更新数据库？这是个经典的两难问题。”

大师：“小白把缓存世界最核心的四个问题全列出来了。这正是我今天要系统梳理的——Redis 在食光集市里的三个角色：高速缓存、分布式锁、限流计数器。”

"缓存模式：旁路缓存（Cache-Aside）——这是最主流的模式：

1. 读：先查 Redis → 命中返回，未命中查 MySQL → 写入 Redis（设 TTL）→ 返回
2. 写：先更新 MySQL → 删除 Redis 对应的 key

这个模式的核心假设是：**缓存是数据库的从属品，数据库永远是正确的。**下次读请求过来，缓存未命中，自动从数据库重建。"

技术映射：Redis 缓存 = 前台接待员——80% 的问题（常客菜单）直接回答（命中缓存），剩下 20% 需要翻档案柜（查数据库），翻完后把答案记在小本子上（回写缓存）。档案柜锁坏了，全公司排队等翻柜（数据库被打爆）。

小胖：“那缓存击穿怎么办？热点 key 过期的那一刻，100 个请求同时打到数据库——这跟没缓存有什么区别？”

小白：“对，还有缓存穿透——如果用户请求了一个不存在的dish_id=-1，缓存和数据库都没有，每次请求都穿透到数据库，恶意攻击可以靠这个打爆数据库。另外——分布式锁的正确用法是什么？我见过一个用 RedisSETNX实现的锁，最后把锁删错了（删了别人的锁）。”

大师：“三个实战场景，逐一拆解。”

"缓存击穿——互斥锁（单飞）模式：热点 key 过期时，只让一个请求去加载数据库，其他请求等待：

defget_menu(store_id):cached=redis.get(f"menu:{store_id}")ifcached:returncached# 获取重建锁——只有一个人去查 DBlock_key=f"lock:menu:{store_id}"ifredis.set(lock_key,"1",nx=True,ex=10):# 获取成功try:data=db.query(...)# 查数据库redis.set(f"menu:{store_id}",data,ex=300)returndatafinally:redis.delete(lock_key)else:# 别人在重建，等一小会儿再查缓存time.sleep(0.1)returnget_menu(store_id)# 递归重试

“缓存穿透——布隆过滤器或空值缓存：对不存在的数据也缓存一个短 TTL 的空标记redis.set(f"dish:{id}", "NULL", ex=60)——避免每次穿透到数据库。”

"分布式锁——Redlock 思路：

# 正确加锁（SET NX PX）locked=redis.set("lock:order:123",unique_value,nx=True,px=30000)# 正确解锁（Lua 脚本保证原子性——只删自己的锁）redis.eval("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) end return 0",1,lock_key,unique_value)

核心要点：用SET resource value NX PX 30000原子加锁 + 设过期时间；解锁用 Lua 脚本验证 value 是否匹配（防止删错锁）；锁必须有过期时间（防止死锁）。"

技术映射：单飞模式 = 厕所门上的"有人"标志——第一个人进去锁门（获取重建锁），后面的人看到标志就在门外等（重试查缓存）。布隆过滤器 = 火车站安检——快速告诉你"这个人肯定不在系统里"，避免在空数据上浪费时间。

小胖：“分布式锁的过期时间设多久？太短了业务没执行完锁就过期了——另一个请求进来以为自己拿到了锁，这叫’锁竞争’吧？”

大师：“对——这就是 Redisson 的watchdog机制要解决的问题：加锁后启动一个后台协程，定期（比如每 10 秒）检查锁是否还被当前客户端持有，如果是就续期（expire 延长）。Python 里可以用threading.Timer或asyncio.create_task实现类似的续期逻辑。”

"另一个常见需求——限流计数器：滑动窗口限流，用 Redis 的 Sorted Set：

now=time.time()window=now-60# 过去 60 秒redis.zremrangebyscore("ratelimit:user:123",0,window)# 清理过期记录count=redis.zcard("ratelimit:user:123")# 当前窗口内请求数ifcount<100:redis.zadd("ratelimit:user:123",{str(now):now})returnTrue# 放行returnFalse# 限流

3. 项目实战：菜单缓存 + 分布式锁 + 限流

环境准备

mkdirfoodmarket-ch21&&cdfoodmarket-ch21 python-mvenv .venv .venv\Scripts\activate pipinstallredis pytest

分步实现

步骤1：实现旁路缓存菜单服务

src/cache_service.py：

"""菜单缓存——Cache-Aside 模式 + 击穿防护"""importjsonimporttimeimportloggingfromtypingimportOptionalimportredis logger=logging.getLogger(__name__)# 模拟数据库MOCK_MENU_DB={"ST001":[{"id":"D001","name":"宫保鸡丁","price":28},{"id":"D002","name":"麻婆豆腐","price":18}],"ST002":[{"id":"D003","name":"清蒸鲈鱼","price":68}],}classMenuCacheService:def__init__(self,redis_client:redis.Redis,ttl:int=300):self.redis=redis_client self.ttl=ttldefget_menu(self,store_id:str)->list[dict]:"""获取门店菜单——带缓存击穿防护"""cache_key=f"menu:{store_id}"# 1. 先查缓存cached=self.redis.get(cache_key)ifcached:logger.info(f"[缓存命中]{store_id}")returnjson.loads(cached)# 2. 缓存未命中——获取重建锁（单飞）lock_key=f"lock:menu:{store_id}"lock_acquired=self.redis.set(lock_key,"1",nx=True,ex=10)iflock_acquired:try:logger.info(f"[缓存未命中]{store_id}—— 重建中")data=MOCK_MENU_DB.get(store_id,[])ifdata:self.redis.set(cache_key,json.dumps(data,ensure_ascii=False),ex=self.ttl)else:# 缓存穿透防护——空值也缓存，短 TTLself.redis.set(cache_key,"[]",ex=30)returndatafinally:self.redis.delete(lock_key)else:# 别人在重建，等待后重试logger.info(f"[等待重建]{store_id}")time.sleep(0.05)returnself.get_menu(store_id)# 递归重试definvalidate_menu(self,store_id:str):"""菜单变更时删除缓存"""self.redis.delete(f"menu:{store_id}")logger.info(f"[缓存失效]{store_id}")defwarm_up(self):"""预热——启动时加载热门门店菜单到缓存"""forstore_idinMOCK_MENU_DB:data=MOCK_MENU_DB[store_id]self.redis.set(f"menu:{store_id}",json.dumps(data,ensure_ascii=False),ex=self.ttl)logger.info(f"[缓存预热] 完成{len(MOCK_MENU_DB)}个门店")

步骤2：实现分布式锁 + 限流器

src/lock_service.py：

"""分布式锁 + 限流器——Redis 实现"""importtimeimportuuidimportloggingimportredis logger=logging.getLogger(__name__)# Lua 脚本：原子解锁UNLOCK_SCRIPT=""" if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end """classRedisLock:"""分布式锁——SET NX PX + Lua 原子解锁"""def__init__(self,redis_client:redis.Redis,lock_key:str,expire_seconds:int=30):self.redis=redis_client self.lock_key=lock_key self.expire_seconds=expire_seconds self.lock_value=str(uuid.uuid4())# 唯一标识，防止误删self._acquired=Falsedefacquire(self,timeout:float=0)->bool:"""获取锁——可选的超时等待"""deadline=time.time()+timeoutwhileTrue:self._acquired=self.redis.set(self.lock_key,self.lock_value,nx=True,ex=self.expire_seconds)ifself._acquired:returnTrueiftime.time()>=deadline:returnFalsetime.sleep(0.05)defrelease(self):"""释放锁——仅释放自己持有的锁"""ifnotself._acquired:returnself.redis.eval(UNLOCK_SCRIPT,1,self.lock_key,self.lock_value)self._acquired=Falsedef__enter__(self):ifnotself.acquire():raiseRuntimeError(f"获取锁超时:{self.lock_key}")returnselfdef__exit__(self,*args):self.release()classRateLimiter:"""滑动窗口限流器——Redis Sorted Set 实现"""def__init__(self,redis_client:redis.Redis,max_requests:int=100,window_seconds:int=60):self.redis=redis_client self.max_requests=max_requests self.window_seconds=window_secondsdefis_allowed(self,user_id:str)->bool:"""检查用户是否在限流窗口内"""key=f"ratelimit:{user_id}"now=time.time()window_start=now-self.window_secondswithself.redis.pipeline()aspipe:pipe.zremrangebyscore(key,0,window_start)# 清理过期pipe.zcard(key)# 当前窗口计数pipe.zadd(key,{str(now):now})# 添加本次请求pipe.expire(key,self.window_seconds+10)# 设置 key 过期_,count,_,_=pipe.execute()returncount<self.max_requests

步骤3：模拟秒杀扣减——分布式锁保护

src/seckill.py：

"""秒杀——分布式锁保护库存关键区"""importredisfromsrc.lock_serviceimportRedisLock STOCK_KEY="seckill:stock:D001"definit_stock(r:redis.Redis,quantity:int=100):r.set(STOCK_KEY,str(quantity))defseckill(r:redis.Redis,user_id:str)->bool:"""秒杀扣减——分布式锁保护"""lock=RedisLock(r,f"seckill:lock:D001:{user_id}",expire_seconds=5)ifnotlock.acquire(timeout=1.0):returnFalse# 获取锁超时try:stock=int(r.get(STOCK_KEY)or0)ifstock<=0:returnFalse# 检查是否已购买（防重复）purchased_key=f"seckill:purchased:{user_id}"ifr.exists(purchased_key):returnFalser.decr(STOCK_KEY)r.set(purchased_key,"1",ex=3600)# 标记已购买returnTruefinally:lock.release()

步骤4：编写测试

tests/test_services.py：

importpytestimportredisimporttimeimportthreadingfromsrc.cache_serviceimportMenuCacheServicefromsrc.lock_serviceimportRedisLock,RateLimiterfromsrc.seckillimportinit_stock,seckill@pytest.fixturedefr():"""连接本地 Redis（若无则跳过）"""try:client=redis.Redis(host="localhost",port=6379,db=15,decode_responses=True)client.ping()yieldclient client.flushdb()client.close()exceptredis.ConnectionError:pytest.skip("Redis 未运行")classTestMenuCache:deftest_cache_hit(self,r):svc=MenuCacheService(r,ttl=60)r.set("menu:ST001",'[{"id":"D001","name":"宫保鸡丁","price":28}]')menu=svc.get_menu("ST001")assertmenu[0]["name"]=="宫保鸡丁"deftest_cache_miss_rebuild(self,r):svc=MenuCacheService(r,ttl=60)menu=svc.get_menu("ST001")assertlen(menu)==2assertr.exists("menu:ST001")# 已写入缓存deftest_invalidate(self,r):svc=MenuCacheService(r,ttl=60)svc.warm_up()assertr.exists("menu:ST001")svc.invalidate_menu("ST001")assertnotr.exists("menu:ST001")classTestRedisLock:deftest_lock_and_release(self,r):lock=RedisLock(r,"test:lock",expire_seconds=10)assertlock.acquire()lock.release()# 释放后别人可以再获取lock2=RedisLock(r,"test:lock",expire_seconds=10)assertlock2.acquire()lock2.release()deftest_cannot_acquire_twice(self,r):lock1=RedisLock(r,"test:lock2",expire_seconds=10)assertlock1.acquire()lock2=RedisLock(r,"test:lock2",expire_seconds=10)assertnotlock2.acquire(timeout=0.1)lock1.release()deftest_context_manager(self,r):withRedisLock(r,"test:lock3",expire_seconds=10)aslock:assertlock._acquired# 退出后释放lock2=RedisLock(r,"test:lock3",expire_seconds=10)assertlock2.acquire()lock2.release()classTestSeckill:deftest_successful_purchase(self,r):init_stock(r,10)assertseckill(r,"user1")isTrueassertint(r.get("seckill:stock:D001"))==9deftest_no_stock(self,r):init_stock(r,0)assertseckill(r,"user1")isFalsedeftest_no_duplicate_purchase(self,r):init_stock(r,5)assertseckill(r,"user1")isTrueassertseckill(r,"user1")isFalse# 防重复

运行：

# 确保本地 Redis 在运行（docker run -d -p 6379:6379 redis:7-alpine）python-mpytest tests/-v

完整代码清单

foodmarket-ch21/ ├── src/ │ ├── __init__.py │ ├── cache_service.py # 旁路缓存 + 击穿防护 │ ├── lock_service.py # 分布式锁 + 限流器 │ └── seckill.py # 秒杀示例 ├── tests/ │ └── test_services.py └── requirements.txt

4. 项目总结

优点 & 缺点

适用场景

1. 读多写少的热点数据：菜单、商品详情、用户信息——缓存命中率 > 90%。
2. 跨服务实例的并发控制：分布式锁——秒杀、定时任务互斥、选主。
3. API 限流/防刷：滑动窗口计数器——用户维度、IP 维度、接口维度。
4. 会话管理：分布式 Session 共享——多实例无状态。
5. 消息队列缓存：排序集合做延迟队列——订单超时取消。

不适用场景

1. 强一致性要求的金融交易：缓存是最终一致性模型，余额类的精确扣减应直接走数据库 + 悲观锁。
2. 超大 value（>10MB）：Redis 单 key 过大会阻塞网络传输，改用对象存储。

注意事项

• 缓存 key 命名规范：{业务}:{实体}:{ID}如menu:ST001、lock:order:SG-001。
• 锁超时时间 > 业务执行时间：如果锁在业务执行完之前过期，另一个实例会错误地拿到锁。
• KEYS *命令在生产禁用：O(n) 复杂度且会阻塞 Redis。用SCAN替代。
• Redis 单实例不满足 CAP 的一致性：如果 Redis 挂了，缓存数据全丢（但可重建）。

常见踩坑经验

故障案例1：先删缓存再更新数据库——中间空窗期导致不一致

现象：先redis.delete(key)再db.update(...)，delete 和 update 之间另一个请求读到了旧数据写回了缓存——用户看到的价格一直不更新。
根因：删除缓存和写数据库不是原子的。
修复：先更新数据库，再删除缓存（Cache-Aside 标准做法）；或用延迟双删——更新前删一次，更新后延迟 500ms 再删一次。

故障案例2：分布式锁误删——A 的锁被 B 删掉

现象：A 获取锁后业务执行超时，锁自动过期。B 获取了锁开始执行，A 的业务终于执行完，释放了锁——但此时锁是 B 的！C 又获取了锁，A、B、C 同时执行临界区。
根因：解锁时没验证锁的 value 是否匹配。
修复：用 Lua 脚本原子比较 value 再删除。

故障案例3：热 key 发现太晚——单分片被打爆

现象：大促时menu:ST001成为热 key，所有请求打到同一个 Redis 分片——该分片 CPU 100%，其他分片 < 10%。
根因：集群模式下单 key 只会落在一个 slot。
修复：多副本缓存——menu:ST001:0、menu:ST001:1……随机读一个，更新时全量失效。

思考题

1. 分布式锁的"续期狗"机制是为了防止什么场景？如果不用续期而把过期时间设得非常长（如 10 分钟），会有什么风险？

2. 缓存预热时一次性加载 10000 个门店的菜单到 Redis，流量会突然涌向——是涌向 Redis 还是 MySQL？如何控制预热速率？

答案见中级篇综合实战章附录。

延伸阅读与资源

Python 3实战精进：从脚本到高并发订单引擎
MongoDB 实战进阶与内核修炼
python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析