本文目录导读：
- 目录导读
- 为什么需要分布式锁？—— 场景与问题
- Python + Redis 分布式锁的核心原理
- 从零实现一个可靠的分布式锁（代码实战）
- 常见问题与避坑指南（含问答）
- 性能优化与生产环境建议
- 总结：何时必须用Redis锁？
Python脚本分布式锁用Redis实现吗？一文讲透原理、代码与最佳实践
目录导读
- 为什么需要分布式锁？—— 场景与问题
- Python + Redis 分布式锁的核心原理
- 从零实现一个可靠的分布式锁（代码实战）
- 常见问题与避坑指南（含问答）
- 性能优化与生产环境建议
- 何时必须用Redis锁？
为什么需要分布式锁？—— 场景与问题
在单机Python脚本中，threading.Lock 或 multiprocessing.Lock 已经能解决并发冲突，但在微服务、多进程部署、多机协作的场景下，资源竞争跨越不同进程甚至不同机器,传统锁就失效了。
典型场景：
- 多个Python Worker同时处理同一订单，导致重复扣款
- 定时任务集群中，多个节点同时执行同一批数据迁移
- 缓存重建时，多个请求同时查询数据库导致雪崩
核心问题：如何让分布在不同机器上的Python脚本，对同一资源达成“唯一访问”的共识？
答案之一就是：基于Redis实现的分布式锁。
Python + Redis 分布式锁的核心原理
Redis实现分布式锁的核心思路很简单：
利用Redis的
SETNX（Set if Not eXists）命令，加上过期时间，保证只有一个客户端能成功设置键值。
但“能用”和“可靠”之间，隔着几个大坑,一个成熟的分布式锁必须满足：
- 互斥性：任意时刻只有一个客户端持有锁
- 安全性：锁会被自动释放，不会死锁（过期时间兜底）
- 可重入性：同一客户端可多次获取同一锁（可选,视业务而定）
- 容错性：Redis主从切换时锁不会丢失（需Redlock算法支持）
原理流程图：
客户端A请求锁：
-> SET lock_key random_value NX PX 30000
（NX：不存在才设置，PX：过期时间30秒）
-> 成功返回OK，客户端A获得锁，执行业务
-> 执行完毕后用Lua脚本原子地检查并删除锁
客户端B请求锁：
-> 同一SET命令，因key已存在，返回nil
-> B等待或快速失败重试注意：释放锁不能用简单的
DEL，必须用Lua脚本验证value是否为自己当初设的值,防止误删别人的锁。
从零实现一个可靠的分布式锁（代码实战）
我们先实现一个生产可用的Python Redis分布式锁，核心依赖：redis-py 和 uuid。
1 最简实现（有缺陷，用于对比）
import redis
import uuid
import time
r = redis.Redis()
def acquire_lock(lock_name, timeout=30):
identifier = str(uuid.uuid4())
lock_key = f"lock:{lock_name}"
# 问题：SETNX与EXPIRE不是原子操作
if r.setnx(lock_key, identifier):
r.expire(lock_key, timeout)
return identifier
return None
def release_lock(lock_name, identifier):
lock_key = f"lock:{lock_name}"
# 问题：GET与DEL非原子，可能误删
if r.get(lock_key) == identifier:
r.delete(lock_key)
缺陷：设置锁和设置过期时间非原子；释放锁时非原子,中间可能被其他进程修改。
2 生产级实现（原子操作）
Redis 2.6.12版本后，SET命令支持更丰富的参数,我们一次性完成：
import redis
import uuid
r = redis.Redis(decode_responses=True)
class RedisDistributedLock:
def __init__(self, redis_client, lock_name, expire_seconds=30, retry_seconds=0.1):
self.redis = redis_client
self.lock_key = f"lock:{lock_name}"
self.expire_seconds = expire_seconds
self.retry_seconds = retry_seconds
self.identifier = str(uuid.uuid4())
def acquire(self):
while True:
# 原子操作：SET key random_value NX PX expire_ms
result = self.redis.set(
self.lock_key,
self.identifier,
nx=True,
px=self.expire_seconds * 1000
)
if result:
return True
# 锁被占用，等待重试
time.sleep(self.retry_seconds)
def release(self):
# 使用Lua脚本保证原子性：检查值一致才删除
lua_script = """
if redis.call("GET", KEYS[1]) == ARGV[1] then
return redis.call("DEL", KEYS[1])
else
return 0
end
"""
self.redis.eval(lua_script, 1, self.lock_key, self.identifier)
# 使用示例
lock = RedisDistributedLock(r, "order:123")
if lock.acquire():
try:
# 执行业务代码
print("处理订单123")
finally:
lock.release()
3 使用场景优化：加锁续期（看门狗）
如果业务执行时间超过过期时间，锁会被自动释放，导致其他进程进入，解决方案是：启动一个后台线程,在锁过期前持续续期。
import threading
def _renew(self):
while self.renewing:
# 每10秒续期一次，假设过期时间30秒
self.redis.expire(self.lock_key, self.expire_seconds)
time.sleep(self.expire_seconds / 3)
# 在acquire成功后启动线程，release时终止线程
常见问题与避坑指南（含问答）
Q1：Python脚本中Redis分布式锁可以用Redlock算法吗？
答：对于绝大多数业务（如秒杀、定时任务），单Redis节点的SET NX + 过期时间已经足够，Redlock适用于对一致性要求极高且Redis主从切换频繁的超大规模系统，Python有成熟库redlock-py，但会增加复杂度,请按需选择。
Q2：锁过期时间设多少合适？
答：建议设为一个预期执行时间的2-3倍，比如业务通常1秒完成，设3秒，并配合看门狗续期，过短会误释，过长会阻塞其他请求，生产环境建议设30秒,并允许续期。
Q3：如果Redis挂了怎么办？
答：单点Redis故障会导致所有锁失效,解决方案：
- 使用Redis Sentinel或Cluster提供高可用
- 接受短暂的服务不可用，降级为重试或本地排队
- 对于极端重要的锁，采用Redlock跨多个独立Redis节点
Q4：分布式锁误删其他线程的锁如何预防？
答：必须用Lua脚本（如上文代码）或者用Redis事务WATCH/MULTI/EXEC，核心原理：删除前必须校验value是否为当前线程的标识符。
Q5：Python中redis-py的SET和SETEX有什么区别？
答：SET key value NX PX 1000是组合原子操作，而SETNX+EXPIRE不是。永远不要用非原子组合实现分布式锁。
性能优化与生产环境建议
- 连接池复用：每个Python脚本/进程维护一个Redis连接池，避免每次锁操作都创建连接。
pool = redis.ConnectionPool(host='xxx', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool)
- 减少锁粒度：不要锁整个订单表，而是锁
order:{order_id}，Redis 6.x支持Key散列,高并发下避免热key。 - 超时回退：
acquire方法中的while True在极端情况下会造成无限循环，建议增加最大重试次数：max_attempts = 10 for _ in range(max_attempts): if self.redis.set(...): return True time.sleep(self.retry_seconds) return False - 监控锁占用时间：在Redis中记录锁的创建时间戳,通过慢查询监控异常长期锁。
何时必须用Redis锁？
| 场景 | 推荐方案 |
|---|---|
| 单机多进程 | multiprocessing.Lock（不依赖外部服务） |
| 多机定时任务 | Redis分布式锁（简单可靠） |
| 秒杀扣库存 | Redis锁 + Lua脚本原子扣减（避免回滚） |
| 高一致性金融 | Zookeeper/Etcd锁（强一致性优于Redis） |
问题Python脚本分布式锁用Redis实现吗？
答案是的，对于95%的互联网业务场景，Redis分布式锁是性价比最高、实现最简洁**的方案，但需要注意原子操作、超时续期和误删问题，结合Python的redis-py和Lua脚本,可以在百行代码内构建一个生产级分布式锁。
最终建议
如果你想快速接入，可以直接使用redis-py官方推荐的锁实现（r.lock()，但版本较低的客户端不支持自动续期），对于要求更高的场景，参考本文的看门狗+Lua脚本方案。锁的本质是权衡——安全性和性能之间的平衡。
关键词：Python、Redis、分布式锁、SET NX、Lua脚本、原子操作、看门狗续期
