1. 项目背景业务场景食光集市川味狂欢节活动上线当天中午 12:00流量峰值时刻——大屏监控上 MySQL 的 CPU 从 15% 飙到 98%。DBA 紧急介入发现一个 SQL 的执行次数高达每秒 4000 次SELECT*FROMdishesWHEREstore_idST001ANDis_available1;这是每家门店的菜单查询——北京朝阳店ST001是最大的门店每天承载 60% 的流量。4000 QPS 打到数据库其中 3800 次都是完全重复的查询——菜单每 5 分钟才更新一次但代码里每次首页刷新都要查一次数据库。这不仅是性能的问题——当数据库被打满后支付回调、库存扣减这些真正重要的事务也被排队阻塞。一顿操作下来MySQL 连接池耗尽整个订单系统进入假死状态 12 分钟。与此同时运营想给川味狂欢节增加一个限时秒杀功能——每个人限购一份需要做分布式并发控制。开发团队开始思考用数据库锁不行太慢。用 Python 线程锁不行多台服务器之间互相看不见。答案指向 Redis——一套工具同时解决两个问题缓存加速 分布式锁。痛点没有缓存的数据库直连架构在高并发下会出现数据库被打爆重复查询占满连接池真正的写操作下单、支付无法获取连接用户看到超时 扣款成功但订单未生成。热点数据雪崩缓存过期那一刻大量请求同时穿透到数据库——瞬间流量打到后端比没缓存时更致命。缓存与数据库不一致更新了数据库但忘了删缓存用户看到的价格是旧价格下单后实际扣款是新价格——标价 28 元付款 32 元的投诉。多实例无锁可加Python 的threading.Lock只在同一个进程内有效。部署了 4 个服务实例需要一把让 4 个进程都看到的分布式锁。2. 项目设计场景MySQL CPU 98% 的红色告警大屏下DBA 小马冲进会议室“ST001 的菜单查询一秒 4000 次你们是不是又在 for 循环里查数据库了”小胖“不是不是就是正常的首页菜单查询——只是 ST001 流量太大了。我觉得加个缓存就行了——Redis 里放一份菜单请求来了先从 Redis 查查不到再查 MySQL。这不就简单了”小白“小胖你只说了最简单的情况。缓存世界里有很多’陷阱’——缓存穿透查不存在的数据、缓存击穿热点 key 过期瞬间大量并发查 DB、缓存雪崩大量 key 同时过期。这些不是’加个 Redis 就完事’的问题。而且——缓存和数据库的一致性怎么保证先删缓存还是先更新数据库这是个经典的两难问题。”大师“小白把缓存世界最核心的四个问题全列出来了。这正是我今天要系统梳理的——Redis 在食光集市里的三个角色高速缓存、分布式锁、限流计数器。”缓存模式旁路缓存Cache-Aside——这是最主流的模式读先查 Redis → 命中返回未命中查 MySQL → 写入 Redis设 TTL→ 返回写先更新 MySQL → 删除 Redis 对应的 key这个模式的核心假设是**缓存是数据库的从属品数据库永远是正确的。**下次读请求过来缓存未命中自动从数据库重建。技术映射Redis 缓存 前台接待员——80% 的问题常客菜单直接回答命中缓存剩下 20% 需要翻档案柜查数据库翻完后把答案记在小本子上回写缓存。档案柜锁坏了全公司排队等翻柜数据库被打爆。小胖“那缓存击穿怎么办热点 key 过期的那一刻100 个请求同时打到数据库——这跟没缓存有什么区别”小白“对还有缓存穿透——如果用户请求了一个不存在的dish_id-1缓存和数据库都没有每次请求都穿透到数据库恶意攻击可以靠这个打爆数据库。另外——分布式锁的正确用法是什么我见过一个用 RedisSETNX实现的锁最后把锁删错了删了别人的锁。”大师“三个实战场景逐一拆解。”缓存击穿——互斥锁单飞模式热点 key 过期时只让一个请求去加载数据库其他请求等待defget_menu(store_id):cachedredis.get(fmenu:{store_id})ifcached:returncached# 获取重建锁——只有一个人去查 DBlock_keyflock:menu:{store_id}ifredis.set(lock_key,1,nxTrue,ex10):# 获取成功try:datadb.query(...)# 查数据库redis.set(fmenu:{store_id},data,ex300)returndatafinally:redis.delete(lock_key)else:# 别人在重建等一小会儿再查缓存time.sleep(0.1)returnget_menu(store_id)# 递归重试“缓存穿透——布隆过滤器或空值缓存对不存在的数据也缓存一个短 TTL 的空标记redis.set(fdish:{id}, NULL, ex60)——避免每次穿透到数据库。”分布式锁——Redlock 思路# 正确加锁SET NX PXlockedredis.set(lock:order:123,unique_value,nxTrue,px30000)# 正确解锁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 Setnowtime.time()windownow-60# 过去 60 秒redis.zremrangebyscore(ratelimit:user:123,0,window)# 清理过期记录countredis.zcard(ratelimit:user:123)# 当前窗口内请求数ifcount100:redis.zadd(ratelimit:user:123,{str(now):now})returnTrue# 放行returnFalse# 限流3. 项目实战菜单缓存 分布式锁 限流环境准备依赖版本说明Python3.13.14基准版本redis-py5.2Redis 客户端pytest8.3测试mkdirfoodmarket-ch21cdfoodmarket-ch21 python-mvenv .venv .venv\Scripts\activate pipinstallredis pytest分步实现步骤1实现旁路缓存菜单服务src/cache_service.py菜单缓存——Cache-Aside 模式 击穿防护importjsonimporttimeimportloggingfromtypingimportOptionalimportredis loggerlogging.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:int300):self.redisredis_client self.ttlttldefget_menu(self,store_id:str)-list[dict]:获取门店菜单——带缓存击穿防护cache_keyfmenu:{store_id}# 1. 先查缓存cachedself.redis.get(cache_key)ifcached:logger.info(f[缓存命中]{store_id})returnjson.loads(cached)# 2. 缓存未命中——获取重建锁单飞lock_keyflock:menu:{store_id}lock_acquiredself.redis.set(lock_key,1,nxTrue,ex10)iflock_acquired:try:logger.info(f[缓存未命中]{store_id}—— 重建中)dataMOCK_MENU_DB.get(store_id,[])ifdata:self.redis.set(cache_key,json.dumps(data,ensure_asciiFalse),exself.ttl)else:# 缓存穿透防护——空值也缓存短 TTLself.redis.set(cache_key,[],ex30)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(fmenu:{store_id})logger.info(f[缓存失效]{store_id})defwarm_up(self):预热——启动时加载热门门店菜单到缓存forstore_idinMOCK_MENU_DB:dataMOCK_MENU_DB[store_id]self.redis.set(fmenu:{store_id},json.dumps(data,ensure_asciiFalse),exself.ttl)logger.info(f[缓存预热] 完成{len(MOCK_MENU_DB)}个门店)步骤2实现分布式锁 限流器src/lock_service.py分布式锁 限流器——Redis 实现importtimeimportuuidimportloggingimportredis loggerlogging.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:int30):self.redisredis_client self.lock_keylock_key self.expire_secondsexpire_seconds self.lock_valuestr(uuid.uuid4())# 唯一标识防止误删self._acquiredFalsedefacquire(self,timeout:float0)-bool:获取锁——可选的超时等待deadlinetime.time()timeoutwhileTrue:self._acquiredself.redis.set(self.lock_key,self.lock_value,nxTrue,exself.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._acquiredFalsedef__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:int100,window_seconds:int60):self.redisredis_client self.max_requestsmax_requests self.window_secondswindow_secondsdefis_allowed(self,user_id:str)-bool:检查用户是否在限流窗口内keyfratelimit:{user_id}nowtime.time()window_startnow-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_seconds10)# 设置 key 过期_,count,_,_pipe.execute()returncountself.max_requests步骤3模拟秒杀扣减——分布式锁保护src/seckill.py秒杀——分布式锁保护库存关键区importredisfromsrc.lock_serviceimportRedisLock STOCK_KEYseckill:stock:D001definit_stock(r:redis.Redis,quantity:int100):r.set(STOCK_KEY,str(quantity))defseckill(r:redis.Redis,user_id:str)-bool:秒杀扣减——分布式锁保护lockRedisLock(r,fseckill:lock:D001:{user_id},expire_seconds5)ifnotlock.acquire(timeout1.0):returnFalse# 获取锁超时try:stockint(r.get(STOCK_KEY)or0)ifstock0:returnFalse# 检查是否已购买防重复purchased_keyfseckill:purchased:{user_id}ifr.exists(purchased_key):returnFalser.decr(STOCK_KEY)r.set(purchased_key,1,ex3600)# 标记已购买returnTruefinally:lock.release()步骤4编写测试tests/test_services.pyimportpytestimportredisimporttimeimportthreadingfromsrc.cache_serviceimportMenuCacheServicefromsrc.lock_serviceimportRedisLock,RateLimiterfromsrc.seckillimportinit_stock,seckillpytest.fixturedefr():连接本地 Redis若无则跳过try:clientredis.Redis(hostlocalhost,port6379,db15,decode_responsesTrue)client.ping()yieldclient client.flushdb()client.close()exceptredis.ConnectionError:pytest.skip(Redis 未运行)classTestMenuCache:deftest_cache_hit(self,r):svcMenuCacheService(r,ttl60)r.set(menu:ST001,[{id:D001,name:宫保鸡丁,price:28}])menusvc.get_menu(ST001)assertmenu[0][name]宫保鸡丁deftest_cache_miss_rebuild(self,r):svcMenuCacheService(r,ttl60)menusvc.get_menu(ST001)assertlen(menu)2assertr.exists(menu:ST001)# 已写入缓存deftest_invalidate(self,r):svcMenuCacheService(r,ttl60)svc.warm_up()assertr.exists(menu:ST001)svc.invalidate_menu(ST001)assertnotr.exists(menu:ST001)classTestRedisLock:deftest_lock_and_release(self,r):lockRedisLock(r,test:lock,expire_seconds10)assertlock.acquire()lock.release()# 释放后别人可以再获取lock2RedisLock(r,test:lock,expire_seconds10)assertlock2.acquire()lock2.release()deftest_cannot_acquire_twice(self,r):lock1RedisLock(r,test:lock2,expire_seconds10)assertlock1.acquire()lock2RedisLock(r,test:lock2,expire_seconds10)assertnotlock2.acquire(timeout0.1)lock1.release()deftest_context_manager(self,r):withRedisLock(r,test:lock3,expire_seconds10)aslock:assertlock._acquired# 退出后释放lock2RedisLock(r,test:lock3,expire_seconds10)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-alpinepython-mpytest tests/-v完整代码清单foodmarket-ch21/ ├── src/ │ ├── __init__.py │ ├── cache_service.py # 旁路缓存 击穿防护 │ ├── lock_service.py # 分布式锁 限流器 │ └── seckill.py # 秒杀示例 ├── tests/ │ └── test_services.py └── requirements.txt4. 项目总结优点 缺点方案缓存模式一致性击穿防护学习成本Cache-Aside本章★★★★ 灵活★★★ 最终一致★★★ 需手写★★★★Read-Through★★★★ 自动★★★ 最终一致★★★★★★★ 需写 loaderWrite-Through★★★★★★★ 同步写—★★★Write-Behind★★★★★ 最快★★ 异步延迟—★★适用场景读多写少的热点数据菜单、商品详情、用户信息——缓存命中率 90%。跨服务实例的并发控制分布式锁——秒杀、定时任务互斥、选主。API 限流/防刷滑动窗口计数器——用户维度、IP 维度、接口维度。会话管理分布式 Session 共享——多实例无状态。消息队列缓存排序集合做延迟队列——订单超时取消。不适用场景强一致性要求的金融交易缓存是最终一致性模型余额类的精确扣减应直接走数据库 悲观锁。超大 value10MBRedis 单 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……随机读一个更新时全量失效。思考题分布式锁的续期狗机制是为了防止什么场景如果不用续期而把过期时间设得非常长如 10 分钟会有什么风险缓存预热时一次性加载 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 实战修炼与源码剖析