怎样在Python应用中实现Redis分布式锁防击穿

直接用SET key value EX seconds NX不够安全，因无法防止客户端崩溃后锁残留或业务超时导致锁被误删；正确做法是加锁用唯一value（如UUID），解锁必须通过Lua脚本原子校验value后再删除。

怎样在python应用中实现redis分布式锁防击穿_使用redis-py库

为什么直接用 `SET key value EX seconds NX` 不够安全

很多人以为加个过期时间再加 NX 就是分布式锁，但实际运行中会遇到两个硬伤：一是客户端崩溃导致锁没释放，二是锁过期后业务还没执行完，另一个客户端抢到锁，造成并发写。Redis 官方推荐的正确做法是「唯一标识 + Lua 脚本原子释放」，即每个锁值必须是客户端自己生成的随机 token（比如 UUID），释放时用 Lua 判断 get(key) == token 再删，否则可能误删别人持有的锁。

如何用 `redis-py` 正确实现可重入、防误删的锁

不要自己拼 SET 命令，优先使用 redis-py 3.0+ 提供的 lock() 方法，它已内置 token 生成、自动续期（extend）、Lua 安全释放逻辑。关键参数要设对：

timeout：锁自动过期时间（秒），建议设为业务最长执行时间的 2–3 倍，避免频繁续期
blocking_timeout：阻塞等待锁的最大时间，防止无限卡住
lock_class：如需自定义行为（比如加日志或监控），可继承 redis.lock.Lock

示例：

import redis
import time
<p>r = redis.Redis()
lock = r.lock("order:123", timeout=10, blocking_timeout=3)</p><p>if lock.acquire():
try:</p><h1>执行扣库存等敏感操作</h1><pre class="brush:php;toolbar:false;">    time.sleep(2)
finally:
    lock.release()  # 自动校验 token，非原子 delete

缓存击穿场景下，锁的生命周期怎么和缓存配合

防击穿的核心不是“只加锁”，而是“锁 + 双检 + 空值缓存”。常见错误是：查缓存没命中 → 加锁 → 查 DB → 写缓存 → 释放锁，但中间若 DB 也为空，下次请求还会重复走锁流程。必须在写缓存时一并写入空值（带短过期）：

Python 3.14.2

Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本，属于3.14系列的第二个维护更新。该版本包含了18项修复，重点解决了多进程、数据类及正则表达式等模块的回归问题，并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式（移除GIL）正式获得官方支持，是Python发展的重要里程碑。

下载

DB 查询结果为 None，则写入 cache.set("key", "__NULL__", ex=60)
读缓存时先判断是否等于 "__NULL__"，是则直接返回空，不进锁
锁内查询 DB 后，仍要再次检查缓存是否存在（双检），避免重复写

注意：redis-py 的 lock() 默认不支持自动续期（auto_renewal=True 需手动开启），高耗时操作务必打开，否则锁提前过期会导致击穿失效。

生产环境必须绕开的几个坑

Redis 单点故障、主从异步复制、客户端时钟漂移都会让锁失效。这不是代码能完全解决的，但可以降低风险：

别用哨兵模式下的 Redis 做强一致性锁——主挂了，从升主前可能丢锁；建议用 Redis Cluster 或 RedLock（多实例投票），但后者在 redis-py 中需自行实现
锁 key 必须带业务上下文，比如 "lock:goods:{goods_id}"，不能写死成 "lock:goods"，否则所有商品抢同一把锁
释放锁失败时不抛异常（lock.release() 返回 False），要主动检查返回值并记录告警
本地时间不准会导致 timeout 计算偏差，建议锁逻辑里用 redis.time() 获取服务端时间做基准