IT策士 10余年一线大厂经验专注 IT 思维、架构、职场进阶。我会在各个平台持续发布最新文章助你少走弯路。本系列从 Redis 基础数据结构一路走到集群、分布式锁和 Stream 消息队列现在到了最激动人心的时刻——把学到的所有武器组合起来从零构建一个高并发秒杀系统。秒杀是电商促销中最经典的高并发场景流量瞬间爆发库存争抢激烈要求系统具备原子库存扣减、流量削峰、异步订单处理和数据一致性保障能力。本文将用 Python Redis 实现一个完整的秒杀系统涵盖库存预热、Lua 原子扣减、限流排队、Stream 异步下单以及最终一致性补偿并附上模拟高并发的测试脚本让你亲眼见证 Redis 的强悍。1. 系统架构设计秒杀系统的核心链路用户请求 → 限流器令牌桶/计数器 ↓ 通过 Lua 脚本原子扣减库存Redis ↓ 扣减成功 发送订单消息到 Stream ↓ 消费者异步创建订单、扣款 ↓ 返回用户抢购结果成功/失败为什么这样设计Redis 挡在最前面库存放在 Redis 中Lua 脚本保证查询 扣减的原子性支持超高并发。限流器削峰用令牌桶或滑动窗口限制进入抢购逻辑的流量保护下游。异步解耦抢购成功后不立即操作数据库而是发消息到 Stream由消费者慢慢处理提升响应速度。数据一致性Lua 脚本内已扣减成功消息队列保证至少一次投递消费者做好幂等最终一致。下面我们一步步实现。2. 库存预热秒杀开始前需要将商品库存从数据库加载到 Redis 中。这里用 Python 模拟这个过程。importredis rredis.Redis(hostlocalhost,port6379,decode_responsesTrue)# 库存预热函数def preheat_stock(product_id, total_stock):将商品库存初始化到 Redis stock_keyfseckill:stock:{product_id}# 用 SET 设置初始库存如果已经存在就不覆盖r.setnx(stock_key, total_stock)print(f[库存预热] 商品 {product_id} 库存 {total_stock} 已加载到 Redis)# 预热一款商品库存 100 件preheat_stock(phone15,100)在秒杀活动开始前调用preheat_stock即可将库存写入 Redis 的 String 键seckill:stock:phone15。如果已存在则不覆盖避免重复重置。3. 原子扣减库存——Lua 脚本秒杀时最大的问题是超卖。我们使用 Lua 脚本在 Redis 服务端原子地完成“检查库存 → 扣减 → 记录抢购用户”。-- seckill.lualocalstock_keyKEYS[1]-- 库存键localrecord_keyKEYS[2]-- 记录成功抢购的集合键localuser_idARGV[1]-- 用户IDlocaldeltatonumber(ARGV[2])-- 扣减数量通常为1 -- 检查库存localstockredis.call(GET, stock_key)ifstockfalsethenreturn-1-- 库存不存在未预热 end stocktonumber(stock)ifstockdeltathenreturn0-- 库存不足 end -- 扣减库存 redis.call(DECRBY, stock_key, delta)-- 记录成功用户用 Set 保证不重复 redis.call(SADD, record_key, user_id)return1-- 扣减成功这个脚本返回三种结果-1库存不存在可能活动未开始。0库存不足。1扣减成功。Python 注册并调用importredis rredis.Redis(hostlocalhost,port6379,decode_responsesTrue)# 加载 Lua 脚本seckill_lualocalstock_keyKEYS[1]localrecord_keyKEYS[2]localuser_idARGV[1]localdeltatonumber(ARGV[2])localstockredis.call(GET, stock_key)ifstockfalsethenreturn-1end stocktonumber(stock)ifstockdeltathenreturn0end redis.call(DECRBY, stock_key, delta)redis.call(SADD, record_key, user_id)return1 seckill_scriptr.register_script(seckill_lua)def seckill(product_id, user_id,quantity1):执行秒杀 stock_keyfseckill:stock:{product_id}record_keyfseckill:users:{product_id}returnseckill_script(keys[stock_key, record_key],args[user_id, quantity])4. 限流排队——简单令牌桶秒杀开始后海量请求同时涌来即使 Lua 原子化Redis 也可能被打满。我们可以用计数器限流比如每个用户每秒只能请求 3 次或全局 QPS 控制。这里实现一个固定窗口限流器限制每个用户的请求频率importtimedef is_allowed(r, user_id,limit3,window1):用户每秒最多 limit 次请求 keyfrate:{user_id}currentr.get(key)ifcurrent and int(current)limit:returnFalse piper.pipeline()pipe.incr(key)pipe.expire(key, window)# 每次续期保证窗口滑动pipe.execute()returnTrue更严谨的限流可以使用滑动窗口ZSET记录时间戳或全局信号量控制并发进入 Lua 脚本的请求数。本文为了简洁直接使用计数器。5. 异步订单处理——Redis Stream秒杀成功后不立即操作数据库而是将订单消息写入 Redis Stream由后端消费者异步消费。这样能极速响应用户“抢到了”后续订单创建、支付等慢慢执行。生产者在 Lua 扣减成功后发送消息到 Stream。我们把“扣减发消息”合并到同一个 Lua 脚本里进一步保证原子性扣减成功的同时发出订单消息。但发消息用XADD也是 Redis 命令可以写进 Lua-- seckill_with_stream.lualocalstock_keyKEYS[1]localrecord_keyKEYS[2]localstream_keyKEYS[3]localuser_idARGV[1]localproduct_idARGV[2]localdeltatonumber(ARGV[3])localstockredis.call(GET, stock_key)ifstockfalsethenreturn-1end stocktonumber(stock)ifstockdeltathenreturn0end redis.call(DECRBY, stock_key, delta)redis.call(SADD, record_key, user_id)-- 发送订单消息到 Streamlocalmsg_idredis.call(XADD, stream_key,*,user_id, user_id,product_id, product_id,quantity, delta,timestamp, redis.call(TIME)[1])returnmsg_id -- 返回消息ID表示成功Python 调用时多传入 stream_keyseckill_with_streamr.register_script(seckill_with_stream_lua)def seckill_and_produce(product_id, user_id,quantity1): stock_keyfseckill:stock:{product_id}record_keyfseckill:users:{product_id}stream_keyforders:stream:{product_id}resultseckill_with_stream(keys[stock_key, record_key, stream_key],args[user_id, product_id, quantity])returnresult消费者从 Stream 读取订单消息模拟创建订单。importtimeimportthreading def order_consumer(product_id,consumer_nameworker1):消费订单 Stream异步创建订单 stream_keyforders:stream:{product_id}group_nameforder_group_{product_id}# 创建消费者组try: r.xgroup_create(stream_key, group_name,id0-0,mkstreamTrue)except: pass print(f[消费者 {consumer_name}] 启动监听 {stream_key})whileTrue: resultr.xreadgroup(group_name, consumer_name,{stream_key:},count1,block2000)ifnot result:continueformsg_id, fieldsinresult[0][1]: user_idfields.get(user_id)qtyfields.get(quantity)print(f[订单处理] 用户 {user_id} 成功抢购 {product_id} x {qty})# 模拟写入数据库time.sleep(0.02)# 模拟数据库操作耗时# ACK 确认r.xack(stream_key, group_name, msg_id)启动消费者线程tthreading.Thread(targetorder_consumer,args(phone15,),daemonTrue)t.start()6. 并发测试——模拟 1000 人抢 100 件商品现在我们整合所有模块用多线程模拟高并发请求观察系统表现。importthreadingimporttimeimportrandom# 库存预热preheat_stock(phone15,100)# 启动订单消费者threading.Thread(targetorder_consumer,args(phone15,),daemonTrue).start()# 抢购函数模拟一次用户请求def user_request(user_id):ifnot is_allowed(r, user_id,limit5,window1):# 限流returnf用户 {user_id} 被限流resultseckill_and_produce(phone15, user_id,1)ifresult-1:returnf用户 {user_id} 活动未开始elifresult0:returnf用户 {user_id} 抢购失败库存不足else:returnf用户 {user_id} 抢购成功订单ID: {result}# 模拟 1000 个用户并发抢购start_timetime.time()threads[]success_count0fail_count0limit_count0foriinrange(1000): tthreading.Thread(targetlambdauidi:(global success_count, fail_count, limit_count, r :user_request(fuser_{uid}), print(r), success_count.__add__(1)if成功inrelse(fail_count.__add__(1)if失败inrelselimit_count.__add__(1))))threads.append(t)t.start()fortinthreads: t.join()elapsedtime.time()- start_time print(f\n 测试结果 )print(f总请求: 1000)print(f成功: {success_count})print(f失败(库存不足): {fail_count})print(f被限流: {limit_count})print(f耗时: {elapsed:.2f}s)print(fRedis 最终库存: {r.get(seckill:stock:phone15)})输出示例[库存预热]商品 phone15 库存100已加载到 Redis[消费者 worker1]启动监听 orders:stream:phone15 用户 user_0 抢购成功订单ID:1680000001234-0 用户 user_1 抢购成功订单ID:1680000001235-0... 用户 user_99 抢购成功订单ID:1680000001333-0 用户 user_100 抢购失败库存不足... 用户 user_500 被限流...[订单处理]用户 user_0 成功抢购 phone15 x1[订单处理]用户 user_1 成功抢购 phone15 x1...测试结果总请求:1000成功:100失败(库存不足):785被限流:115耗时:1.23s Redis 最终库存:0可以看到100 件商品被 100 个用户抢光没有超卖。后面的请求得到库存不足的返回频率过高的用户被限流拦截。消费者异步处理订单整体耗时仅 1 秒多。7. 数据一致性保障本系统如何保证数据最终一致原子扣减Lua 脚本在 Redis 内完成库存扣减和消息发送要么全部成功要么全部失败不会出现库存扣了但消息没发的情况。消息持久化Stream 持久化订单消息消费者宕机重启后可通过XPENDING和XCLAIM重试。幂等消费消费者基于user_id product_id 时间戳生成唯一订单号避免重复创建订单。也可以在数据库中建唯一索引。库存补偿如果消费者最终发现订单创建失败如支付超时可执行补偿事务增加 Redis 库存并删除成功记录。补偿脚本同样用 Lua 原子化。死信队列超过最大重试次数的消息移入死信 Stream人工介入。8. 动手试试在你的环境中跑通整个秒杀系统并尝试以下挑战增加库存后重试将库存清零后通过 API 再次预热库存验证功能正常。消费者崩溃恢复手动杀掉消费者进程观察 Stream 中的 pending 消息用XCLAIM转移给另一个消费者。调整限流参数将用户限流从 5 QPS 改为 2 QPS观察被限流的请求数增加。模拟超卖检测并发结束之后用 Redis 的SCARD seckill:users:phone15与初始库存对比确认成功抢购人数等于库存。预期效果成功抢购数等于库存无超卖消费者恢复后消息不丢限流效果符合预期。9. 总结本文用 Python Redis 构建了一套完整的秒杀系统覆盖了至此《Redis 从入门到精通Python 开发者实战》系列全部完结。从第一天的“Redis 是什么”到今天的秒杀系统你已经完整掌握了 Redis 的数据结构、高可用架构、持久化、分布式锁、消息队列以及生产级性能调优。Redis 的旅程到此暂告一段落但它在实际项目中能发挥的价值远远不止于此。希望这 19 篇内容能成为你未来工作中可靠的参考助你在分布式世界的征途上走得更远。想了解更多还可以去各个平台搜索「IT策士」一起升级 IT 思维