小张是个独立开发者，自己搭了个二手球鞋交易网站。每次限量款发售，他都得提前喝两杯咖啡盯着后台。因为一秒内涌入上千人抢十双鞋。数据库瞬间卡死，卖出了15双——库存成了负数。他下定决心要重写秒杀程序。

从场景还原来看。库存为什么变成负数

先看小张原来的代码：

def create_order: product = Product.query.get if product.stock> 0: product.stock -= 1 db.session.commit create_order_record return "成功" return "已售罄"

看起来没问题，但高并发下藏着巨大的坑。当两个请求同时读到product.stock = 1都判断stock> 0成立。接下来各自减1提交，库存就从1变成了-1。

问题的根源在于“读-判断-写”不是原子操作。不过，多个请求交错执行，互相看不见对方正要做的修改。老实说，

再看方案一。数据库行锁

用数据库的行锁机制，让更新操作变成串行执行。MySQL的InnoDB引擎在更新时会自动锁住这一行。

def create_order_with_lock: from sqlalchemy import text # 用原生SQL加FOR UPDATE。锁住这一行 sql = text product = db.session.execute.fetchone if product.stock> 0: update_sql = text db.session.execute db.session.commit create_order_record return "成功" db.session.rollback return "已售罄"

FOR UPDATE让第一个拿到锁的事务执行完之前，其他所有请求都在排队等待。这样库存肯定减不超，但性能直线下降——每个请求都得等前一个提交才能继续。

适合并发量不大的场景。小张的网站高峰时上千人抢，这么搞数据库连接池会瞬间爆满。

从方案二来看，Redis预扣库存

把库存从数据库搬到内存里。Redis单线程处理命令，天然就没有并发问题。

import redis import json r = redis.Redis def init_seckill: """秒杀活动开始前，把库存存入Redis""" r.set # 记录已下单的使用者，防止重复抢 r.delete def seckill: # 用Lua脚本保证原子性 lua_script = """ local stock_key = KEYS local users_key = KEYS local user_id = ARGV -- 检查是否已经抢过 if redis.call == 1 n return -1 end # 扣库存 local stock = redis.call if stock>= 0 n redis.call return stock else # 库存不足。回滚 return -1 end""" # 改进版：先检查再扣 lua_fixed = """ local stock_key = KEYS local users_key = KEYS local user_id = ARGV if redis.call == 1 n return -2 end local stock = redis.call if not stock or tonumber <= 0 n return -1 end redis.call redis.call return tonumber - 1""" stock_key = f"stock:{product_id}" users_key = f"users:{product_id}" result = r.eval if result == -2: return "您已经抢过了" elif result == -1: return "已售罄" else: # 异步写入数据库 async_save_order return f"抢到了剩余{result}件"

Lua脚本在Redis里是原子执行的，整个过程不会被其他命令打断。decr之前先get检查库存，彻底杜绝超卖。

从方案三来看。消息队列削峰填谷

Redis扛住了抢购请求，但每个成功使用者都要写数据库创建订单。上万请求同时写数据库，照样会崩。

用消息队列把写操作变成异步的。使用者点击抢购后立刻返回“排队中”，后台慢慢处理。

import pika import threading from queue import Queue import time# 简单的内存队列 order_queue = Queue def async_save_order: 至于"""生产者，把订单放入队列""" order_queue.put({ "user_id": user_id,"product_id": product_id,"timestamp": time.time }) def order_worker: """使用者这方面。后台线程慢慢写数据库 while True: order_data = order_queue.get 从try来看，# 这里写数据库 create_order_record print except Exception as e: print # 失败重试 time.sleep order_queue.put finally: order_queue.task_done#启动4个线程并发 for _ in range: t = threading.Thread t.start

实际环境会用RabbitMQ或Kafka。使用者按数据库能承受的速度慢慢，秒杀瞬间的流量洪就被削了。

至于极致调整。令牌桶限流

即便Redis再好，也不可能无限 加上限流，保护程序不被刷垮。

import time class TokenBucket: def __init__: self.rate = rate # 每秒补充令牌数 self.capacity = capacity # 桶的容量 self.tokens = capacity self.last_refill = time.time def acquire: now = time.time # 补充令牌 elapsed = now - self.last_refill self.tokens = min self.last_refill = now if self.tokens>= 1: self.tokens -= 1 return True return False#每个商品独立限流器。每秒只放行100个请求 limiter = TokenBucketdef seckill_with_limit: if not limiter.acquire: return "程序繁忙，请稍后再试" return seckill

令牌桶比计数器算法更平滑。漏桶强行让请求匀速通过令牌桶允许短时间突发流量——比如前.5秒用掉100个令牌，后0.5秒就只能等令牌慢慢补充。

说到完整实战，从开始到结束

把上面所有组件拼起来形成一个完整的秒杀流程：

from flask import Flask,request。jsonify import redis import threading import time app = Flask r = redis.Redis#初始化秒杀活动def init_seckill: r.set r.set r.delete#调整的Lua脚本SECILL_LUA = """ stock = redis.call if not stock or tonumber <= 0 n return -1 end user_exists = redis.call if user_exists == 1 n return -2 end redis.call redis.call return tonumber - 1"""@app.routedef seckill_api: user_id = request.args.get if not user_id: return jsonify stock_key = f"stock:{product_id}" users_key = f"users:{product_id}" result = r.eval if result == -1: return jsonify elif result == -2: return jsonify else: # 异步落库 order_queue.put return jsonifyif __name__ == '__main__': init_seckill #商品1号，库存10件 app.run

用wrk或ab压测一下：

#模拟200个并发，总共1000个请求 wrk -t4 -c200 -d10s --timeout=2s "http://localhost:/seckill/?user_id=1"

Redis轻松扛住几千并发。库存始终没超卖，数据库订单表也因为有队列保护而稳如泰山。

踩坑经验分享

Redis挂了怎么办？说起来，秒杀开始前做一次库存全量备份到数据库。Redis宕机时快速降级到数据库行锁方案，虽然慢但不会丢失订单。

使用者重复点击怎么办？前端按钮置灰只做一半工作。真正防重靠Lua脚本里的sember检查。更彻底的办法是用SETNX给每个使用者加一个短时锁：

def prevent_double_click: lock_key = f"click_lock:{user_id}:{product_id}" if r.setnx: r.expire #2秒过期 return True return False

库存热key问题：几千人抢同一个商品，Redis单节点网可能被打满。可以用本地缓存分摊压力：

from cachetools import TTache local_cache = TTache #1秒过期 def get_stock_with_cache: if product_id in local_cache: return local_cache stock = r.get local_cache = stock return stock

每个服务器缓存1秒，把Redis的查询压力降低几十倍。