分布式锁是分布式系统中协调并发访问共享资源的核心机制。在微服务架构下，库存扣减、订单防重、定时任务调度等场景均依赖分布式锁保证数据一致性。本文对比Redis和ZooKeeper两种主流分布式锁实现方案，分析其优劣势并给出代码实现。
分布式锁的核心要求
一个可靠的分布式锁需要满足四个基本条件：
– 互斥性：同一时刻只有一个客户端持有锁
– 避免死锁：锁必须设置过期时间，防止持有者崩溃后锁无法释放
– 容错性：部分节点故障时锁服务仍可用
– 可重入性：同一客户端可多次获取同一把锁（可选，视场景需求）
Redis单节点分布式锁：SET NX EX方案
Redis单节点分布式锁利用SET命令的NX（不存在才设置）和EX（过期时间）选项实现原子加锁。
import redis
import uuid
import time
class RedisLock:
def __init__(self, redis_client, lock_key, expire_time=10):
self.redis = redis_client
self.lock_key = lock_key
self.expire_time = expire_time
self.lock_value = str(uuid.uuid4())
def acquire(self, retry_interval=0.1, max_retries=50):
for _ in range(max_retries):
result = self.redis.set(
self.lock_key,
self.lock_value,
nx=True,
ex=self.expire_time
)
if result:
return True
time.sleep(retry_interval)
return False
def release(self):
# 使用Lua脚本保证原子性：检查值匹配后删除
script = '''
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
'''
return self.redis.eval(script, 1, self.lock_key, self.lock_value)
def __enter__(self):
if not self.acquire():
raise Exception("Failed to acquire lock")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()关键设计点：lock_value使用UUID防止误删其他客户端的锁。释放锁时通过Lua脚本验证持有者身份再删除，保证检查和删除的原子性。如果先GET判断再DEL，两步之间存在时间窗口，其他客户端可能在此期间获取锁导致误删。
可重入锁实现
import redis
import threading
import time
class RedisReentrantLock:
def __init__(self, redis_client, lock_key, expire_time=30):
self.redis = redis_client
self.lock_key = f"reentrant:{lock_key}"
self.expire_time = expire_time
self.thread_local = threading.local()
def acquire(self):
if not hasattr(self.thread_local, 'count'):
self.thread_local.count = 0
if self.thread_local.count > 0:
self.thread_local.count += 1
return True
lock_value = str(uuid.uuid4())
self.thread_local.lock_value = lock_value
for _ in range(100):
if self.redis.set(self.lock_key, lock_value, nx=True, ex=self.expire_time):
self.thread_local.count = 1
return True
time.sleep(0.1)
return False
def release(self):
if self.thread_local.count > 1:
self.thread_local.count -= 1
return
script = '''
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
end
return 0
'''
self.redis.eval(script, 1, self.lock_key, self.thread_local.lock_value)
self.thread_local.count = 0可重入锁使用ThreadLocal记录当前线程的重入次数，同一线程多次调用acquire时计数加1，release时减1，归零时才真正释放锁。适用于递归调用和嵌套场景。
Redis RedLock算法：多节点容错方案
RedLock由Redis作者Antirez提出，通过多个独立Redis节点组成锁集群，客户端向N个节点同时申请锁，获得多数（N/2+1）节点成功即认为加锁成功。假设5个节点，需至少3个节点成功。
import redis
import time
import uuid
class RedLock:
def __init__(self, redis_nodes, lock_key, expire_time=10):
self.nodes = [redis.Redis(**node) for node in redis_nodes]
self.lock_key = lock_key
self.expire_time = expire_time # 单位：毫秒
self.lock_value = str(uuid.uuid4())
self.quorum = len(self.nodes) // 2 + 1
def acquire_instance(self, node):
try:
return node.set(
self.lock_key,
self.lock_value,
nx=True,
px=self.expire_time
)
except redis.RedisError:
return False
def release_instance(self, node):
script = '''
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
end
return 0
'''
try:
node.eval(script, 1, self.lock_key, self.lock_value)
except redis.RedisError:
pass
def acquire(self, max_retry=3):
for _ in range(max_retry):
success_count = 0
start_time = time.time()
for node in self.nodes:
if self.acquire_instance(node):
success_count += 1
elapsed = (time.time() - start_time) * 1000
remaining = self.expire_time - elapsed
if success_count >= self.quorum and remaining > 0:
return True
else:
for node in self.nodes:
self.release_instance(node)
time.sleep(0.05)
return False
def release(self):
for node in self.nodes:
self.release_instance(node)
# 使用示例
nodes = [
{"host": "192.168.1.10", "port": 6379},
{"host": "192.168.1.11", "port": 6379},
{"host": "192.168.1.12", "port": 6379},
{"host": "192.168.1.13", "port": 6379},
{"host": "192.168.1.14", "port": 6379},
]
lock = RedLock(nodes, "order_lock:12345", expire_time=10000)
if lock.acquire():
try:
process_order()
finally:
lock.release()RedLock的核心假设是各Redis节点独立部署、无主从复制，避免单点故障和主从切换导致的锁丢失问题。elapsed时间用于校准锁的实际有效期，防止网络延迟导致锁已过期但客户端不知情。
ZooKeeper分布式锁：临时顺序节点方案
ZooKeeper通过临时顺序节点（Ephemeral Sequential Node）实现分布式锁，利用Watch机制实现公平锁排队，天然具备锁失效自动清理能力。
from kazoo.client import KazooClient
from kazoo.exceptions import NodeExistsError, NoNodeError
import time
class ZkLock:
def __init__(self, zk_hosts, lock_path, timeout=30):
self.zk = KazooClient(hosts=zk_hosts, timeout=timeout)
self.lock_path = lock_path
self.lock_node = None
self.lock_name = "lock-"
def connect(self):
self.zk.start()
self.zk.ensure_path(self.lock_path)
def acquire(self):
# 创建临时顺序节点
self.lock_node = self.zk.create(
f"{self.lock_path}/{self.lock_name}",
ephemeral=True, sequence=True
)
while True:
children = self.zk.get_children(self.lock_path)
children.sort()
my_seq = self.lock_node.split('/')[-1]
my_index = children.index(my_seq)
if my_index == 0:
return True
prev_node = f"{self.lock_path}/{children[my_index - 1]}"
event = self.zk.handler.event_object()
def on_deleted(e):
event.set()
if self.zk.exists(prev_node, watch=on_deleted):
event.wait(timeout=30)
def release(self):
if self.lock_node:
try:
self.zk.delete(self.lock_node)
except NoNodeError:
pass
self.lock_node = None
def disconnect(self):
self.zk.stop()
self.zk.close()
# 使用示例
zk_lock = ZkLock("192.168.1.20:2181", "/locks/order")
zk_lock.connect()
if zk_lock.acquire():
try:
process_order()
finally:
zk_lock.release()
zk_lock.disconnect()ZooKeeper锁的核心机制：每个客户端创建一个临时顺序节点，序号最小者获得锁。未获锁的客户端监听前一个节点的删除事件，前一个节点释放锁后唤醒下一个客户端。临时节点在客户端Session断开时自动删除，避免死锁。
方案对比与选型建议
Redis单节点锁：实现简单，性能最高（单次操作微秒级），适合单Redis实例环境。缺点是Redis主从切换可能导致锁丢失。
Redis RedLock：多节点容错，安全性高于单节点方案，但实现复杂度高，存在争议（Martin Kleppmann曾指出RedLock在时钟漂移场景下的安全性问题）。适合对性能要求高且能容忍极端情况下锁失效的场景。
ZooKeeper锁：强一致性，Session超时自动释放锁，支持公平锁排队，可靠性最高。缺点是ZooKeeper部署维护成本高，性能低于Redis（毫秒级操作）。适合对数据一致性要求极高的场景。
选型建议：电商库存扣减等高并发场景优先选用Redis锁；金融交易、配置变更等强一致性场景选用ZooKeeper锁。生产环境中还应考虑Etcd分布式锁作为第三种方案，Etcd基于Raft协议实现强一致性，性能介于Redis和ZooKeeper之间，Kubernetes已在生产环境中广泛验证其可靠性。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/fen-bu-shi-suo-shi-xian-fang-an-dui-bi-redisredlock-yu/
