Redis 从入门到精通：分布式锁 —— 从 SETNX 到 Redlock-程序员充电站

IT策士 10余年一线大厂经验，专注 IT 思维、架构、职场进阶。我会在各个平台持续发布最新文章，助你少走弯路。

分布式系统中，多个服务实例常常需要竞争同一资源——比如扣减库存、分配订单号、更新公共配置。如果没有协调机制，并发操作就会导致数据错乱。这时候就需要分布式锁：一把所有实例都认可的“钥匙”，谁拿到谁才能操作资源，操作完再还回去。

Redis 是实现分布式锁最流行的工具之一。但它并非一把SETNX就万事大吉——死锁、误删、锁过期、主从切换导致的锁丢失……这些都是生产环境里的真实大坑。本文从单实例的简单锁开始，一步步演化到生产级的 Redlock 算法，用 Python 把每个坑都踩一遍，再把它填平。

1. 分布式锁的本质与基本要求

一把合格的分布式锁需要满足：

互斥性：同一时刻只有一个客户端持有锁。
防死锁：即使持有锁的客户端崩溃，锁也能被释放（通过过期时间）。
解铃还须系铃人：只有持锁的客户端才能释放锁，不能误删别人的锁。
高可用：锁服务本身不能是单点，否则服务宕机锁就全部失效。
容错性：在部分节点故障时仍能正确加锁、解锁。

Redis 单线程、高性能、自带过期，天然适合做锁。但要满足上述全部要求，需要经过精心设计。

2. 第一代：SETNX —— 朴素的起点

最原始的想法：SETNX（SET if Not eXists），键不存在则设置成功，表示获取锁；键已存在则失败，表示锁被别人占用。释放锁就是DEL键。

# 终端 A127.0.0.1:6379>SETNX lock:order:1001 A(integer)1# 加锁成功# 终端 B（同时）127.0.0.1:6379>SETNX lock:order:1001 B(integer)0# 加锁失败，锁已被 A 持有

Python 版本：

importredis r=redis.Redis(host='localhost',port=6379,decode_responses=True)def acquire_lock_v1(lock_name, holder_id):"""第一代：SETNX 加锁"""returnr.setnx(lock_name, holder_id)def release_lock_v1(lock_name):"""第一代：直接 DEL 释放""" r.delete(lock_name)# 使用ifacquire_lock_v1('lock:order:1001','client_A'): print('A 获取锁成功')# 执行业务...release_lock_v1('lock:order:1001')else: print('A 获取锁失败')

致命问题：

死锁：如果客户端 A 在DEL之前崩溃，锁永远无法释放。其他客户端永远拿不到锁。
误删：客户端 B 无法区分锁是谁的，可能直接DEL掉 A 持有的锁。

3. 第二代：加过期时间 + 持有者标识

3.1 防止死锁——给锁加 TTL

用SET key value EX seconds NX（Redis 2.6.12+）一条命令完成加锁并设置过期：

def acquire_lock_v2(lock_name, holder_id,expire=30):"""第二代：加锁带过期时间"""returnr.set(lock_name, holder_id,nx=True,ex=expire)# 使用ifacquire_lock_v2('lock:order:1001','client_A',expire=30): print('A 获取锁成功，30 秒后自动释放')

即使客户端崩溃，锁也会在 30 秒后自动释放，不会死锁。

3.2 防止误删——释放时校验持有者

释放前先检查锁的值是否与自己的holder_id匹配。但“检查 + 删除”两步不是原子的，需要 Lua 脚本：

release_lock_lua="""ifredis.call("GET", KEYS[1])==ARGV[1]thenreturnredis.call("DEL", KEYS[1])elsereturn0end""" release_lock=r.register_script(release_lock_lua)def acquire_and_release(): lock_name='lock:order:1001'holder='client_A'ifr.set(lock_name, holder,nx=True,ex=30): print(f'{holder} 获取锁成功')# 业务逻辑...result=release_lock(keys=[lock_name],args=[holder])print(f'释放结果: {result}')# 1 表示成功删除else: print('获取锁失败')acquire_and_release()

还有问题吗？有。如果业务执行时间超过了锁的过期时间（30 秒），锁自动释放了，其他客户端拿到锁，原客户端还在操作，就破坏了互斥性。这就需要锁续期。

4. 第三代：锁续期（Watchdog）

如果业务还在执行，锁快过期了，就自动延长过期时间。这就是Watchdog（看门狗）机制。Redisson（Java）中就有经典的看门狗实现。

我们在 Python 中用一个后台线程来实现：

importthreadingimporttimeimportuuid class RedisLockWithWatchdog:"""带看门狗的分布式锁""" def __init__(self, redis_client, lock_name,expire=30,renew_interval=None): self.redis=redis_client self.lock_name=lock_name self.holder=str(uuid.uuid4())self.expire=expire self.renew_interval=renew_interval or max(expire //3,1)# 每 expire/3 秒续期self._renew_thread=None self._stop_renew=threading.Event()def acquire(self):"""获取锁"""ifself.redis.set(self.lock_name, self.holder,nx=True,ex=self.expire): self._start_watchdog()returnTruereturnFalse def _start_watchdog(self):"""启动看门狗线程，定期续期""" self._stop_renew.clear()self._renew_thread=threading.Thread(target=self._watchdog_loop,daemon=True)self._renew_thread.start()def _watchdog_loop(self):"""续期循环""" renew_script=self.redis.register_script("""ifredis.call("GET", KEYS[1])==ARGV[1]thenreturnredis.call("EXPIRE", KEYS[1], ARGV[2])elsereturn0end""")whilenot self._stop_renew.wait(self.renew_interval): result=renew_script(keys=[self.lock_name],args=[self.holder, self.expire])ifresult: print(f'[看门狗] 续期成功，延长 {self.expire}s')else: print('[看门狗] 锁已不属于自己，停止续期')breakdef release(self):"""释放锁""" self._stop_renew.set()ifself._renew_thread: self._renew_thread.join(timeout=2)release_script=self.redis.register_script("""ifredis.call("GET", KEYS[1])==ARGV[1]thenreturnredis.call("DEL", KEYS[1])elsereturn0end""")returnrelease_script(keys=[self.lock_name],args=[self.holder])def __enter__(self):ifself.acquire():returnself raise Exception('获取锁失败')def __exit__(self, exc_type, exc_val, exc_tb): self.release()# 使用示例r=redis.Redis(host='localhost',port=6379,decode_responses=True)def process_order(): try: with RedisLockWithWatchdog(r,'lock:order:1001',expire=10)as lock: print('获取锁成功，开始处理订单...')time.sleep(15)# 模拟长业务，超过 expireprint('订单处理完成')except Exception as e: print(e)process_order()

输出：

获取锁成功，开始处理订单...[看门狗]续期成功，延长 10s[看门狗]续期成功，延长 10s 订单处理完成

即使业务执行了 15 秒（超过初始的 10 秒 TTL），锁也没有过期，因为看门狗在第 3 秒和第 6 秒续期了。

5. 第四代：Redlock —— 分布式锁的巅峰

5.1 单实例锁的致命缺陷

即使加了 Watchdog，上述方案仍有一个根本性缺陷：锁存储在单个 Redis 实例上，一旦这个实例宕机，锁就丢失了。

如果使用主从架构，主节点宕机后从节点提升，但从节点可能还没同步到锁的数据，导致锁丢失，两个客户端同时认为拿到了锁。

Redlock是 Redis 作者 antirez 提出的算法，在多个独立的 Redis 节点上同时加锁，只要超过半数节点加锁成功，才认为获取锁成功。

5.2 Redlock 算法流程

获取当前时间戳（毫秒）。
依次向 N 个 Redis 节点（建议 5 个）请求加锁，使用相同的key、value和 TTL。
客户端设置一个超时时间（远小于锁的 TTL），如果某个节点没及时响应，则立即尝试下一个节点。
统计加锁成功的节点数。如果成功数 ≥ N/2 + 1（即大多数），且总耗时 < 锁的 TTL，则认为获取锁成功。
锁的有效时间 = 锁的 TTL - 总耗时。
如果加锁失败（未达到大多数或耗时过长），则向所有节点发送释放锁请求。

客户端 │ ├── 加锁请求 ──>Redis-1 ✅ ├── 加锁请求 ──>Redis-2 ✅ ├── 加锁请求 ──>Redis-3 ✅ ├── 加锁请求 ──>Redis-4 ❌(超时)└── 加锁请求 ──>Redis-5 ✅ │4/5 成功>半数 → 获取锁成功

5.3 Python Redlock 实现

我们使用redlock-py库（也可手写，这里展示原理）：

importredlockimporttime# 配置 5 个独立的 Redis 节点（生产环境需要不同机器）# 这里演示：可在不同端口启动多个 Redisnodes=[{'host':'localhost','port':6379,'db':0},{'host':'localhost','port':6380,'db':0},{'host':'localhost','port':6381,'db':0},{'host':'localhost','port':6382,'db':0},{'host':'localhost','port':6383,'db':0},]# 创建 Redlock 实例dlm=redlock.Redlock(nodes,retry_count=3,retry_delay=0.2)# 加锁lock_name='lock:critical:task'my_lock=None try:# 尝试获取锁，TTL=10000ms (10s)my_lock=dlm.lock(lock_name,10000)ifmy_lock: print(f'获取 Redlock 成功! 锁值: {my_lock.resource}')# 执行业务...time.sleep(2)else: print('获取 Redlock 失败')except redlock.MultipleRedlockException as e: print(f'Redlock 异常: {e}')finally:ifmy_lock: dlm.unlock(my_lock)print('锁已释放')

手写 Redlock 核心逻辑（理解原理）：

importuuidimporttimeimportredis class SimpleRedlock:"""简化版 Redlock 实现""" def __init__(self, nodes,retry_count=3,retry_delay=0.2):""" nodes:[{'host':'localhost','port':6379,'db':0},...]""" self.nodes=[redis.Redis(host=n['host'],port=n['port'],db=n.get('db',0),decode_responses=True,socket_timeout=0.5)forninnodes]self.quorum=len(self.nodes)//2+1self.retry_count=retry_count self.retry_delay=retry_delay def lock(self, resource,ttl_ms=10000):"""获取分布式锁""" value=str(uuid.uuid4())forattemptinrange(self.retry_count +1): start_time=int(time.time()*1000)success_count=0# 向所有节点尝试加锁fornodeinself.nodes: try:ifnode.set(resource, value,nx=True,px=ttl_ms): success_count+=1except Exception as e: print(f'节点 {node} 加锁异常: {e}')elapsed=int(time.time()*1000)- start_time# 加锁成功条件：多数节点成功，且耗时未超 TTLifsuccess_count>=self.quorum and elapsed<ttl_ms:return{'value':value,'validity':ttl_ms - elapsed# 剩余有效时间}# 加锁失败，释放已成功的节点self._unlock_all(resource, value)# 等待后重试ifattempt<self.retry_count: time.sleep(self.retry_delay *(attempt +1))# 退避returnNone def unlock(self, resource, value):"""释放锁""" self._unlock_all(resource, value)def _unlock_all(self, resource, value):"""向所有节点发送释放锁请求""" unlock_script="""ifredis.call("GET", KEYS[1])==ARGV[1]thenreturnredis.call("DEL", KEYS[1])elsereturn0end"""fornodeinself.nodes: try: node.eval(unlock_script,1, resource, value)except Exception: pass# 忽略释放失败（节点可能已宕机）# 使用 SimpleRedlock# 需要先启动 5 个 Redis 实例（不同端口）# 为演示，这里假设已存在redlock=SimpleRedlock(nodes,retry_count=2)lock_info=redlock.lock('lock:critical:task',ttl_ms=5000)iflock_info: print(f'获取锁成功，剩余有效时间: {lock_info["validity"]}ms')# 业务处理...time.sleep(1)redlock.unlock('lock:critical:task', lock_info['value'])print('锁已释放')else: print('获取锁失败')

6. Redlock 争议与最佳实践

Redlock 并非毫无争议。分布式系统专家 Martin Kleppmann 曾撰文指出 Redlock 存在安全性问题。核心争议在于：时钟跳跃和GC 停顿可能导致锁失效。

时钟跳跃：如果某个节点的系统时钟被 NTP 调整，可能导致锁提前过期。
GC 停顿：客户端在 GC（垃圾回收）时暂停，锁过期了业务还在执行。

反方建议：使用ZooKeeper或etcd这类基于共识算法的强一致性存储来实现锁。

实际业界选择：

大多数互联网公司仍广泛使用 Redis 做分布式锁（简单、高性能、已有基础设施）。
对一致性要求极高的场景（金融交易、订单状态机），建议使用 ZooKeeper/etcd。
折中方案：使用redlock-py库并配合单调递增的 fencing token，每次获取锁返回一个递增的 token，资源服务可以通过比对 token 来拒绝过期的锁。

7. 动手试试

Watchdog 模拟：设置锁过期 5 秒，业务睡眠 12 秒，观察看门狗续期效果。打印续期次数和最终锁释放状态。
并发竞争：启动 10 个线程同时竞争一个锁，统计获取成功的次数（应为 1），验证互斥性。
Redlock 节点故障：用 Docker 启动 5 个 Redis，运行 SimpleRedlock，手动停掉其中 2 个节点，验证仍然可以加锁（3/5 > 半数）。
锁误删防护：用两个不同holder_id尝试释放对方的锁，验证 Lua 脚本正确拦截。