直接用数据库行锁会超卖，因select_for_update()锁持有时间长且无法隔离非DB操作；Redis分布式锁粒度更细、不依赖事务，配合自动续期和原子扣减可有效防超卖。
为什么直接用数据库行锁在高并发下还会超卖
因为 Django 的 select_for_update() 在事务提交前才释放锁，但实际业务中常存在「查库存→判断是否足够→扣减→写入」的多步操作。中间可能有其他请求完成整个流程，导致两次都判断成功。更关键的是，如果事务里混了非数据库操作（比如调用外部 API、发消息），锁持有时间被拉长，吞吐量断崖下跌。
Redis 分布式锁能将锁粒度精确到「某个商品 ID」，且不依赖数据库事务生命周期，适合这种短时、高频、跨进程的互斥场景。
用 redis-py 实现可重入、带自动续期的锁
别手写 SET NX PX —— 容易丢锁或死锁。推荐用 redis-py 官方支持的 Lock 类，它内置唯一 token、看门狗自动续期、可重入计数：
import redis from django.conf import settings <p>r = redis.Redis.from_url(settings.REDIS_URL) lock = r.lock(f"stock_lock:1001", timeout=10, blocking_timeout=3)</p><h1>timeout：锁自动过期时间（秒），防止客户端崩溃后锁不释放</h1><h1>blocking_timeout：阻塞等待锁的最大时间，避免无限卡住</h1>
- 锁 key 必须带业务标识（如
"stock_lock:{product_id}"），不能全局限一个 key -
timeout建议设为「最长单次扣减逻辑耗时 × 2」，太短易误释放，太长影响恢复速度 - 务必检查
lock.acquire()返回值，False表示没抢到锁，应快速失败或降级（比如返回「稍后再试」）
Django 视图中安全扣库存的完整链路
核心是「锁 → 查缓存库存 → 扣减 → 更新缓存 + DB → 释放锁」，所有读写必须在锁内完成，且缓存与 DB 要保持最终一致：
立即学习“Python免费学习笔记（深入）”；
def deduct_stock(request):
product_id = request.POST.get("product_id")
qty = int(request.POST.get("qty", 1))
<pre class='brush:python;toolbar:false;'>lock_key = f"stock_lock:{product_id}"
lock = r.lock(lock_key, timeout=10, blocking_timeout=2)
if not lock.acquire():
return JsonResponse({"error": "busy"}, status=429)
try:
# 1. 从 Redis 读当前库存（不是 DB！）
stock_key = f"stock:{product_id}"
current_stock = r.get(stock_key)
if current_stock is None:
# 缓存未命中，从 DB 加载并回填（注意：这里要加锁保护加载过程，但通常用双检+setnx）
with transaction.atomic():
p = Product.objects.select_for_update().get(id=product_id)
r.set(stock_key, p.stock, ex=60)
current_stock = str(p.stock)
if int(current_stock) < qty:
return JsonResponse({"error": "out of stock"}, status=400)
# 2. 扣减 Redis 库存（原子操作）
new_stock = r.decrby(stock_key, qty)
if new_stock < 0:
# 理论上不会进这里，但防错：回滚 Redis 并抛异常
r.incrby(stock_key, qty)
raise ValueError("stock race condition")
# 3. 异步更新 DB（或放 Celery），避免阻塞锁
update_stock_in_db.delay(product_id, -qty)
return JsonResponse({"success": True, "left": new_stock})
finally:
try:
lock.release()
except redis.exceptions.LockError:
pass # 锁已被自动过期，忽略- 不要在锁内做
Product.objects.get()后再扣减——DB 查询慢，放大锁竞争 -
r.decrby()是原子的，但必须确保初始库存已写入 Redis（可用 cache-aside 模式预热） - DB 更新建议异步，否则锁持有时间包含网络 IO，极易成为瓶颈
容易被忽略的三个致命细节
线上出问题往往卡在这几个点：
- Redis 连接池没配对：Django 多线程/ASGI 下，
redis.Redis实例必须复用连接池，否则每请求新建连接会打爆 Redis（配置connection_pool=redis.ConnectionPool(...)） - 锁 key 没做标准化：商品 ID 若含特殊字符（如冒号、空格），需
urllib.parse.quote编码，否则 Redis 报错或 key 冲突 - 没处理主从同步延迟：若用 Redis 哨兵或集群，写主从异步复制，
r.get()可能读到旧值。解决方案是强制读主节点（r.execute_command("READONLY OFF")）或改用 Redlock（但多数场景没必要）
分布式锁不是银弹，它把问题从「DB 行锁竞争」转移到「Redis 网络延迟与可靠性」。压测时重点看锁获取失败率和 Redis P99 延迟，而不是只盯着 Python 代码执行时间。
