100件库存卖出去130件：Flask秒杀Redis预扣库存加Lua原子扣减

⚠️ 免责声明：本文项目仅供技术学习参考，严禁直接作为毕业设计或课程作业提交，更不得用于任何未授权的真实系统测试。所有演示均在本人本地项目上进行。学术诚信、合法合规从我做起 🙂

先说结论

毕设想加分，加个秒杀最划算。老师一眼就能看出你碰过高并发。

可它也是翻车最狠的地方。我第一次压测，100 件库存卖出 130 件。那多出来的 30 单，不是发货赔钱，就是退款挨骂。

原因特别土：我先查库存，再扣库存，中间隔了几毫秒。并发一起来，10 个请求同时看到"还剩 1 件"，10 个人全扣成功。

这篇我按实际踩坑的顺序写——先让超卖现形，再一层层往上加方案，最后讲库存怎么对账。代码可以直接抄进你的项目。

一、先让问题暴露出来

不压测，你永远不知道自己多脆。我用的就是最土的 ab：

# 200 并发，总共 1000 次请求
ab -n 1000 -c 200 http://127.0.0.1:5000/seckill/1

第一版代码长这样：

@app.route('/seckill/<int:goods_id>', methods=['POST'])
@login_required
def seckill(goods_id):
goods = Goods.query.get_or_404(goods_id)

# ❌ 查
if goods.stock <= 0:
return jsonify({'code': 1, 'message': '已售罄'})

# ❌ 中间隔了一段时间，别人也查到了同样的库存
time.sleep(0.001) # 模拟业务耗时，真实项目里是查优惠券、算价格

# ❌ 扣
goods.stock -= 1
db.session.commit()

order = OrderInfo(user_id=current_user.id, goods_id=goods_id, ...)
db.session.add(order)
db.session.commit()

return jsonify({'code': 0})

跑完一看库：stock 变 -30，订单表躺着 130 条。

这就是超卖。 根子不在代码写错，在"查"和"扣"是两件事，中间留了个窗口，并发请求全从这儿钻进来。

二、第一层：把判断和扣减塞进一条 SQL

最省事的修法，是让数据库自己保证原子性：

@app.route('/seckill/<int:goods_id>', methods=['POST'])
@login_required
def seckill(goods_id):
# ✅ 判断和扣减在同一个 UPDATE 里完成
affected = db.session.execute(
text("UPDATE goods SET stock = stock – 1 WHERE id = :gid AND stock > 0"),
{'gid': goods_id}
).rowcount

if affected == 0:
db.session.rollback()
return jsonify({'code': 1, 'message': '已售罄'})

# 扣成功才建订单
order = OrderInfo(user_id=current_user.id, goods_id=goods_id, ...)
db.session.add(order)
db.session.commit()

return jsonify({'code': 0})

关键是 WHERE stock > 0。数据库执行 UPDATE 时会给这一行加锁，后面的请求得等前一个提交完才能读，读到的就是扣过之后的库存。rowcount 是 0，说明没扣到，直接返回售罄。

再压一次：库存 0，订单 100 条。 超卖没了。

答辩可以这么说：把"读-改-写"三个动作压成一条原子语句，用数据库的行锁保证互斥，消掉检查与执行之间的竞态窗口。

但这条方案有个硬伤：所有请求都打在数据库上。200 并发的时候 MySQL 连接池直接打满，响应时间从 20ms 飙到 3 秒。库存是准了，服务也快挂了。

三、第二层：把库存搬到 Redis

秒杀是典型的读多写少、瞬时高并发。这种活数据库干不好，Redis 在行。

思路就一句：活动开始前把库存预热进 Redis，请求来了先在 Redis 里扣，扣成功再异步落库。数据库从"每秒几百次"掉到"每秒几十次"。

预热

def preload_stock(goods_id):
"""活动开始前调用，把数据库库存写进 Redis"""
goods = Goods.query.get(goods_id)
if not goods:
return False
redis_client.set(f'seckill:stock:{goods_id}', goods.stock)
# 用户集合，用来防重复下单，活动结束记得删
redis_client.delete(f'seckill:users:{goods_id}')
return True

但只用 Redis 也不安全

很多人觉得"Redis 是单线程的，怎么会超卖"，然后写出这样的代码：

# ❌ 这样写照样超卖
stock = int(redis_client.get(f'seckill:stock:{gid}') or 0)
if stock <= 0:
return '售罄'
redis_client.decr(f'seckill:stock:{gid}')

Redis 单线程，说的是"单条命令"原子执行，不是"多条命令"整体原子。 GET 和 DECR 之间，其他客户端的命令照样能插进来。所以这条看着没毛病的代码，压测照样超卖。

四、第三层：Lua 脚本，把多步压成一步

Redis 跑 Lua 脚本时，整个脚本被当成一个整体执行，中间插不进其他客户端的任何命令。这正是我们要的原子性。

顺手把"防重复下单"也塞了进去。

脚本

新建 lua/seckill.lua：

— KEYS[1] = 库存 key
— KEYS[2] = 已购用户集合 key
— ARGV[1] = 当前用户 ID

local stock_key = KEYS[1]
local user_key = KEYS[2]
local user_id = ARGV[1]

— 1. 已经买过就直接拒绝
if redis.call('SISMEMBER', user_key, user_id) == 1 then
return –2
end

— 2. 库存不足
local stock = tonumber(redis.call('GET', stock_key) or '0')
if stock <= 0 then
return –1
end

— 3. 原子扣减 + 记录用户
redis.call('DECR', stock_key)
redis.call('SADD', user_key, user_id)

return stock – 1

脚本返回三个值，调用方好判断：

• >= 0：成功，返回剩余库存
• -1：售罄
• -2：重复下单

Python 调用

import redis
from flask import current_app

# 全局只注册一次，别每次请求都读文件
_seckill_script = None

def get_seckill_script():
global _seckill_script
if _seckill_script is None:
with open('lua/seckill.lua', 'r', encoding='utf-8') as f:
_seckill_script = current_app.redis.register_script(f.read())
return _seckill_script

@app.route('/seckill/<int:goods_id>', methods=['POST'])
@login_required
def seckill(goods_id):
script = get_seckill_script()

result = script(
keys=[f'seckill:stock:{goods_id}', f'seckill:users:{goods_id}'],
args=[current_user.id]
)

if result == –1:
return jsonify({'code': 1, 'message': '已售罄'})
if result == –2:
return jsonify({'code': 1, 'message': '每人限购一件'})

# Redis 扣成功，异步落库
push_order_task(current_user.id, goods_id)
return jsonify({'code': 0, 'message': '抢购成功，正在生成订单'})

def push_order_task(user_id, goods_id):
"""把下单任务丢进队列，由后台 worker 消费"""
current_app.task_queue.put({
'user_id': user_id,
'goods_id': goods_id,
'ts': time.time()
})

register_script 会把脚本内容算成 SHA1，之后每次只发 EVALSHA，省带宽也更快。这是 redis-py 自带的事，不用你操心。

再压：库存 0，订单 100 条，响应时间 30ms 左右，数据库 QPS 从 800 掉到 40。

五、第四层：异步落库和库存回滚

Redis 扣成功只是占了个位，订单还得真写进数据库。这一步挪到后台：

# 后台 worker（毕设用线程就够，生产上 Celery / RQ）
def order_worker():
while True:
task = task_queue.get()
try:
with app.app_context():
# 数据库层面再扣一次，保证两边最终一致
affected = db.session.execute(
text("UPDATE goods SET stock = stock – 1 "
"WHERE id = :gid AND stock > 0"),
{'gid': task['goods_id']}
).rowcount

if affected == 0:
# 数据库已经没库存了，把 Redis 的名额还回去
redis_client.incr(f'seckill:stock:{task["goods_id"]}')
redis_client.srem(f'seckill:users:{task["goods_id"]}',
task['user_id'])
continue

order = OrderInfo(
user_id=task['user_id'],
goods_id=task['goods_id'],
status=0,
expire_at=datetime.now() + timedelta(minutes=15)
)
db.session.add(order)
db.session.commit()
except Exception as e:
app.logger.error(f'落库失败: {e}')
# 失败要还名额，否则库存越扣越少
redis_client.incr(f'seckill:stock:{task["goods_id"]}')
redis_client.srem(f'seckill:users:{task["goods_id"]}',
task['user_id'])
finally:
task_queue.task_done()

有件事常被忘掉：抢到却 15 分钟没付款，库存得还回去。

12 篇那个定时任务正好能复用，只是在回滚数据库库存之外，多还一次 Redis：

def cancel_expired_orders():
now = datetime.now()
expired = OrderInfo.query.filter(
OrderInfo.status == 0,
OrderInfo.expire_at < now
).all()

for order in expired:
order.status = 4
db.session.execute(
text("UPDATE goods SET stock = stock + 1 WHERE id = :gid"),
{'gid': order.goods_id}
)
# ✅ 关键：Redis 的名额也要还
redis_client.incr(f'seckill:stock:{order.goods_id}')
redis_client.srem(f'seckill:users:{order.goods_id}', order.user_id)

db.session.commit()

只还数据库不还 Redis，Redis 里的库存只会越跑越少，最后明明有货也卖不出去。

六、上线前一定要跑对账

Redis 和 MySQL 是两套存储，异步必然有对不齐的时候。毕设可以简化，但得留个对账的口子：

@app.route('/admin/seckill/reconcile/<int:goods_id>')
@admin_required
def reconcile(goods_id):
db_stock = Goods.query.get(goods_id).stock
redis_stock = int(redis_client.get(f'seckill:stock:{goods_id}') or 0)
sold = redis_client.scard(f'seckill:users:{goods_id}')

return jsonify({
'db_stock': db_stock,
'redis_stock': redis_stock,
'sold': sold,
'diff': db_stock – redis_stock,
'ok': db_stock == redis_stock
})

diff 不是 0，说明中间有落库失败没还名额，或者定时任务漏还了。平时挂个定时任务跑一遍，比出事后翻日志强。

七、还有两个容易被老师问到的点

先说限流。秒杀接口不限流，一个人写个循环就能把库存全占光。最省事的写法：

def check_rate_limit(user_id, limit=5, window=10):
key = f'rate:seckill:{user_id}'
count = redis_client.incr(key)
if count == 1:
redis_client.expire(key, window)
return count <= limit

再说别把 Redis 当成唯一真相。库存的权威数据永远在数据库，Redis 只是挡住流量的前置闸门。这个定位得讲明白——答辩老师问"Redis 挂了怎么办"，你有答案：降级回数据库那条 UPDATE … WHERE stock > 0，慢，但不超卖。

答辩高频题

1. 为什么不能只在数据库层做防超卖？

答：能防住超卖，但扛不住并发。所有请求都落在同一行上争锁，连接池很快打满，响应时间从毫秒级涨到秒级。秒杀的核心矛盾是瞬时流量远大于数据库的处理能力，所以要在数据库前面加一层 Redis 挡流量。

2. Redis 是单线程的，为什么还会超卖？

答：单线程保证的是单条命令原子执行，不是多条命令整体原子。GET 判断库存和 DECR 扣减是两条命令，中间其他客户端的命令可以插进来，两个请求都读到"还剩 1 件"就会都扣成功。用 Lua 脚本或者 DECR 之后判断返回值才能解决。

3. Lua 脚本为什么能保证原子性？

答：Redis 执行脚本时会把整个脚本当成一个不可分割的操作，脚本运行期间不会处理其他客户端的命令。所以脚本里的"判断库存 + 扣减 + 记录用户"三步对外部完全不可见，等价于一次原子操作。

4. Redis 扣成功但数据库写失败怎么办？

答：把名额还回 Redis——库存 INCR 加回去，用户从已购集合里移除。同时记录失败日志，用对账接口定期比对 Redis 和数据库的库存差，diff 不为 0 就告警。异步系统只能做到最终一致，关键是让不一致可被观测。

5. 怎么防止一个人抢多个名额？

答：在 Lua 脚本里用 Set 记录已购用户，SISMEMBER 判断，SADD 记录，两步都在同一个脚本里完成，和其他操作互斥。这样同一个用户 ID 第二次进来直接返回 -2，不会走到扣库存那一步。前提是登录态可靠，别让用户能伪造 ID。

写在最后

秒杀这个功能，最见一个人对"并发"的理解。

单机测试时，怎么写都能跑通——查库存、扣库存、建订单，顺得很。

可流量一上来，差别就露出来了。有人用一条 SQL 防住了超卖，却扛不住流量；有人做到"Redis 挡流量 + Lua 保原子 + 异步落库 + 对账兜底"，库存稳、响应快。

差的不在代码行数，在你想没想过那句：如果两个请求同时到达，会发生什么？

想明白这句，比这个功能本身值钱。

梅雅达编程笔记原创 · 首发 CSDN · 转载请注明出处