分布式锁是微服务架构中控制共享资源访问的基础组件。从简单的Redis SETNX到Paxos共识算法，不同方案在一致性保证、性能开销和容错能力上各有取舍。本文对比Redis单节点锁、Redlock算法、Zookeeper锁和etcd锁的实现原理与工程实践。
Redis单节点分布式锁与可靠性问题
Redis分布式锁的最简实现使用SET key value NX PX命令，NX保证互斥性，PX设置过期时间防止死锁。但单节点Redis在主从故障切换时存在锁丢失风险。
import redis
import uuid
import time
class RedisDistributedLock:
def __init__(self, redis_client, lock_key, expire_seconds=30):
self.redis = redis_client
self.lock_key = lock_key
self.expire_seconds = expire_seconds
self.lock_value = str(uuid.uuid4())
def acquire(self, retry_count=3, retry_delay=0.2):
for attempt in range(retry_count):
result = self.redis.set(
self.lock_key, self.lock_value,
nx=True, px=int(self.expire_seconds * 1000)
)
if result:
return True
time.sleep(retry_delay)
return False
def release(self):
release_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
return self.redis.eval(release_script, 1, self.lock_key, self.lock_value)
def __enter__(self):
if not self.acquire():
raise Exception(f"Failed to acquire lock: {self.lock_key}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
# 使用示例
r = redis.Redis(host='localhost', port=6379, db=0)
with RedisDistributedLock(r, "order:lock:1001", expire_seconds=10):
process_order(1001)单节点Redis锁的缺陷：主从异步复制场景下，客户端A在Master获取锁后，锁信息尚未同步到Slave时Master宕机，Sentinel提升Slave为新Master，客户端B获取同一把锁成功，导致互斥性被破坏。
Redlock算法多节点仲裁机制
Redis作者Antirez提出的Redlock算法通过多节点仲裁解决单点故障问题。核心思路：在N个独立Redis实例上同时获取锁，超过半数成功即认为锁获取成功。
import redis
import uuid
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
class Redlock:
def __init__(self, redis_nodes, retry_count=3, retry_delay=0.2):
self.clients = [
redis.Redis(host=h, port=p, socket_timeout=0.1)
for h, p in redis_nodes
]
self.quorum = len(redis_nodes) // 2 + 1
self.retry_count = retry_count
self.retry_delay = retry_delay
def acquire_instance(self, client, lock_key, lock_value, expire_ms):
try:
return client.set(lock_key, lock_value, nx=True, px=expire_ms)
except Exception:
return False
def release_instance(self, client, lock_key, lock_value):
release_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
try:
client.eval(release_script, 1, lock_key, lock_value)
except Exception:
pass
def acquire(self, lock_key, expire_seconds=10):
lock_value = str(uuid.uuid4())
expire_ms = int(expire_seconds * 1000)
for attempt in range(self.retry_count):
start_time = time.time()
success_count = 0
with ThreadPoolExecutor(max_workers=len(self.clients)) as executor:
futures = [
executor.submit(
self.acquire_instance,
client, lock_key, lock_value, expire_ms
)
for client in self.clients
]
for future in as_completed(futures):
if future.result():
success_count += 1
elapsed = (time.time() - start_time) * 1000
validity_time = expire_ms - elapsed
if success_count >= self.quorum and validity_time > 0:
return {'value': lock_value, 'validity_time': validity_time}
for client in self.clients:
self.release_instance(client, lock_key, lock_value)
time.sleep(self.retry_delay)
return None
def release(self, lock_key, lock_info):
for client in self.clients:
self.release_instance(client, lock_key, lock_info['value'])
# 使用 - 5个独立Redis实例
nodes = [
('redis1.example.com', 6379),
('redis2.example.com', 6379),
('redis3.example.com', 6379),
('redis4.example.com', 6379),
('redis5.example.com', 6379),
]
redlock = Redlock(nodes)
lock = redlock.acquire("inventory:deduct:sku_1001", expire_seconds=10)
if lock:
try:
check_and_deduct_inventory("sku_1001")
finally:
redlock.release("inventory:deduct:sku_1001", lock)Zookeeper分布式锁与Watch机制
Zookeeper通过临时顺序节点和Watch机制实现分布式锁，天然具备会话超时自动释放和公平排队特性。与Redis的主动过期不同，Zookeeper锁依赖session心跳维持。
from kazoo.client import KazooClient
from kazoo.exceptions import NoNodeError
import threading
class ZkDistributedLock:
def __init__(self, hosts, lock_path, timeout=10):
self.zk = KazooClient(hosts=hosts, timeout=timeout)
self.lock_path = lock_path
self.node_path = None
def connect(self):
self.zk.start()
def acquire(self, blocking=True, timeout=None):
self.zk.ensure_path(self.lock_path)
self.node_path = self.zk.create(
f"{self.lock_path}/lock-",
ephemeral=True,
sequence=True
)
node_name = self.node_path.split('/')[-1]
while True:
children = self.zk.get_children(self.lock_path)
children.sort()
my_index = children.index(node_name)
if my_index == 0:
return True
prev_node = f"{self.lock_path}/{children[my_index - 1]}"
event = threading.Event()
@self.zk.DataWatch(prev_node)
def watch_node(data, stat):
if stat is None:
event.set()
if self.zk.exists(prev_node):
event.wait(timeout=timeout)
if not event.is_set():
self.zk.delete(self.node_path)
return False
def release(self):
if self.node_path:
try:
self.zk.delete(self.node_path)
except NoNodeError:
pass
self.node_path = None
def close(self):
self.zk.stop()
self.zk.close()
# 使用示例
zk_lock = ZkDistributedLock("zk1:2181,zk2:2181,zk3:2181", "/locks/payment")
zk_lock.connect()
if zk_lock.acquire(timeout=5):
try:
process_payment()
finally:
zk_lock.release()
zk_lock.close()分布式锁方案选型对比与工程建议
三种主流方案的横向对比：
维度 Redis单节点 Redlock Zookeeper etcd
一致性 弱(主从异步) 多数派仲裁 ZAB共识(CP) Raft共识(CP)
性能 最优(~10k QPS) 良好(~1k QPS) 中等(~500 QPS) 良好(~1k QPS)
可用性 AP AP偏向 CP CP
自动续期 需手动实现 需手动实现 Session心跳自动 Lease自动续期
公平性 非公平 非公平 公平(顺序节点) 非公平
故障恢复 锁可能丢失 少数节点故障可容忍 Leader故障短暂不可 Leader故障短暂不可
部署复杂度 低 高(需5+独立实例) 中(3-5节点集群) 中(3-5节点集群)
适用场景 缓存防击穿 资金交易 配置分发/选主 服务发现/配置管理工程实践中，分布式锁的选择应遵循”够用即可”原则：
缓存防击穿、限流计数等容忍极小概率不一致的场景，使用Redis单节点锁即可，配合watchdog自动续期机制（如Redisson的LockWatchdogTimeout）避免业务执行超时导致锁过期。
资金扣减、库存扣减等强一致性要求的场景，优先选择基于Raft/ZAB共识的etcd或Zookeeper锁。Redlock在网络分区下的安全性存在争议（Martin Kleppmann的批评），在极端情况下仍可能出现锁失效，不建议用于最严格的资金场景。
所有分布式锁都应配合数据库唯一约束作为兜底方案。分布式锁解决的是”减少冲突”，数据库约束解决的是”最终正确性”。两层防护才能在分布式环境下实现既高性能又正确的并发控制。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/fen-bu-shi-suo-shi-xian-fang-an-dui-bi-yu-redlock-suan-fa/
