分布式锁在存储系统中的选型与实践Redis Redlock与ZooKeeper方案的深度对比一、一个定时任务在两个节点上同时执行——重复扣款事故的起源定时对账任务是存储系统中最常见的分布式锁应用场景每天凌晨 2 点系统需要扫描过去 24 小时的所有交易记录计算账户余额。这个任务必须保证只在一个节点上执行否则会导致重复扣款。在一次事故中团队使用 RedisSETNX实现了简单锁但网络抖动导致锁过期后两个节点同时获取了锁——结果是 12 个账户被重复扣款金额达 32000 元。同样是分布式锁Redis Redlock 算法和 ZooKeeper 临时顺序节点方案提供了不同的保证强度但在网络分区、时钟漂移、GC 暂停等异常场景下它们的表现差异巨大。理解这些差异是生产环境中选型的前提。二、两种方案的协议原理与行为差异sequenceDiagram participant C1 as Client 1br/(锁持有者) participant C2 as Client 2br/(竞争者) participant R1 as Redis A participant R2 as Redis B participant R3 as Redis C Note over C1: Redlock 加锁流程 C1-R1: SET lock resource value NX PX 30000 C1-R2: SET lock resource value NX PX 30000 C1-R3: SET lock resource value NX PX 30000 R1--C1: OK R2--C1: OK R3--C1: OK Note over C1: 获得锁 (3/3 成功) C1-C1: 执行业务逻辑br/(假设耗时 25s) Note over C2: 锁过期前无法获取 C2-R1: SET lock resource value NX PX 30000 R1--C2: FAIL (key exists) Note over C1: 业务完成, 释放锁 C1-R1: DEL lock (lua check) C1-R2: DEL lock (lua check) C1-R3: DEL lock (lua check) Note over C2: Client 2 获得锁 C2-R1: SET lock resource value NX PX 30000 C2-R2: SET lock resource value NX PX 30000 C2-R3: SET lock resource value NX PX 30000 R1--C2: OK R2--C2: OK R3--C2: OKRedlock 算法核心原理获取当前时间戳 T1依次向 N 个独立的 Redis 节点推荐 5 个请求加锁使用相同的 key 和随机 value计算获取锁消耗的时间 T2 - T1只有当从 N/2 1 个多数派节点成功获取锁且总耗时 锁的有效时间才算获取成功锁的有效时间 原始 TTL - (T2 - T1)如果获取失败向所有节点发送解锁请求ZooKeeper 临时顺序节点方案所有竞争者在同一路径下创建临时顺序节点序号最小的节点获得锁其他节点对前一个节点注册 Watcher锁持有者释放锁删除节点或 Session 超时客户端断开下一个节点自动获得锁三、完整工程实现对比3.1 Redis Redlock 实现import time import uuid import redis from typing import List, Optional from dataclasses import dataclass dataclass class LockResult: acquired: bool lock_value: Optional[str] valid_until: Optional[float] class RedisRedlock: Redis Redlock 算法的生产级实现 CLOCK_DRIFT_FACTOR 0.01 # 时钟漂移系数 def __init__(self, redis_clients: List[redis.Redis]): 初始化 Redlock Args: redis_clients: 至少3个独立的 Redis 连接推荐5个 if len(redis_clients) 3: raise ValueError(Redlock 至少需要 3 个独立的 Redis 实例) self.clients redis_clients self.quorum len(redis_clients) // 2 1 def acquire(self, resource: str, ttl_ms: int 30000, retry_count: int 3) - LockResult: 获取分布式锁 Args: resource: 锁的资源标识 ttl_ms: 锁的过期时间毫秒 retry_count: 重试次数 Returns: LockResult: 加锁结果 lock_value str(uuid.uuid4()) for attempt in range(retry_count): start_time time.monotonic() # 阶段 1尝试在所有节点加锁 acquired_count 0 for client in self.clients: try: if self._lock_instance(client, resource, lock_value, ttl_ms): acquired_count 1 except redis.RedisError as e: # 个别节点故障不影响整体只要达到 quorum continue elapsed_ms (time.monotonic() - start_time) * 1000 # 阶段 2判断是否成功 valid_ttl ttl_ms - elapsed_ms - int(ttl_ms * self.CLOCK_DRIFT_FACTOR) if acquired_count self.quorum and valid_ttl 0: valid_until time.monotonic() valid_ttl / 1000 return LockResult( acquiredTrue, lock_valuelock_value, valid_untilvalid_until ) # 阶段 3加锁失败释放已获取的锁 self._unlock_all(resource, lock_value) # 随机延迟后重试避免惊群效应 if attempt retry_count - 1: time.sleep(0.01 * (1 attempt)) return LockResult(acquiredFalse, lock_valueNone, valid_untilNone) def release(self, resource: str, lock_value: str) - bool: 释放分布式锁 return self._unlock_all(resource, lock_value) def _lock_instance(self, client: redis.Redis, resource: str, lock_value: str, ttl_ms: int) - bool: 在单个 Redis 实例上尝试加锁 try: result client.set( resource, lock_value, nxTrue, pxttl_ms ) return result is True or result OK except redis.RedisError: return False def _unlock_all(self, resource: str, lock_value: str) - bool: 使用 Lua 脚本安全释放锁防止释放别人的锁 unlock_script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end success False for client in self.clients: try: result client.eval(unlock_script, 1, resource, lock_value) if result: success True except redis.RedisError: continue return success def extend(self, resource: str, lock_value: str, additional_ttl_ms: int) - bool: 延长锁的过期时间续期 extend_script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(pexpire, KEYS[1], ARGV[2]) else return 0 end extended 0 for client in self.clients: try: result client.eval( extend_script, 1, resource, lock_value, additional_ttl_ms ) if result: extended 1 except redis.RedisError: continue return extended self.quorum3.2 ZooKeeper 实现from kazoo.client import KazooClient from kazoo.recipe.lock import Lock from kazoo.exceptions import LockTimeoutException import threading class ZKDistributedLock: 基于 ZooKeeper 的分布式锁封装 def __init__(self, zk_hosts: str, root_path: str /locks): 初始化 ZooKeeper 分布式锁 Args: zk_hosts: ZK 集群地址如 zk1:2181,zk2:2181,zk3:2181 root_path: 锁的根路径 self.zk KazooClient(hostszk_hosts) self.zk.start() self.root_path root_path self.zk.ensure_path(root_path) def acquire(self, resource: str, timeout: float 30.0) - bool: 获取分布式锁 Args: resource: 资源名称 timeout: 等待超时时间秒 Returns: bool: 是否成功获取锁 lock_path f{self.root_path}/{resource} self.lock self.zk.Lock(lock_path, identifierthreading.current_thread().name) try: return self.lock.acquire(timeouttimeout) except LockTimeoutException: return False def release(self) - bool: 释放分布式锁 if hasattr(self, lock) and self.lock.is_acquired: self.lock.release() return True return False def close(self): 关闭 ZK 连接 self.zk.stop() self.zk.close() # 使用示例 — 自动续期的锁看门狗 class WatchDogLock: 带看门狗的锁自动续期防止业务执行超时导致锁提前释放 def __init__(self, lock_manager, resource: str, ttl: int 30, renew_interval: int 10): self.lock_mgr lock_manager self.resource resource self.ttl ttl self.renew_interval renew_interval self._watchdog_thread None self._stop_watchdog threading.Event() def __enter__(self): self.lock_mgr.acquire(self.resource, self.ttl) self._start_watchdog() return self def __exit__(self, exc_type, exc_val, exc_tb): self._stop_watchdog.set() if self._watchdog_thread: self._watchdog_thread.join(timeout2) self.lock_mgr.release() def _start_watchdog(self): 启动看门狗线程定期续期锁 def renew_loop(): while not self._stop_watchdog.is_set(): self._stop_watchdog.wait(self.renew_interval) if not self._stop_watchdog.is_set(): try: self.lock_mgr.extend(self.resource, self.ttl) except Exception: # 续期失败记录日志但不中断业务 pass self._watchdog_thread threading.Thread( targetrenew_loop, daemonTrue ) self._watchdog_thread.start()四、方案对比与适用场景维度Redis RedlockZooKeeper 临时节点安全性网络分区有争议Martin Kleppmann 论证存在安全漏洞安全Session 机制保证性能加锁延迟~2-5ms跨 5 个节点~10-50ms涉及 ZAB 协议自动释放时间到期可能过早Session 断开心跳可靠公平性非公平竞争公平FIFO 顺序通知机制无需要轮询Watcher 自动通知部署复杂度需要独立 Redis 集群需要 ZK 集群适用场景高性能、短持有、可容忍小概率误放强一致性要求、需要公平性Redlock 的安全争议Martin Kleppmann 在其论文中指出Redlock 在客户端 GC 暂停超过锁 TTL 时会出现安全性问题——两个客户端可能同时认为自己持有锁。虽然这种场景在实践中极其罕见概率 0.001%但对于金融对账等零容忍场景ZooKeeper 方案更合适。锁持有时间的黄金法则业务逻辑执行时间 锁 TTL 的 70%使用 Redlock 无看门狗业务逻辑执行时间不可预测使用 ZK 看门狗或 Session 自动续期需要对锁持有者做健康检查必须使用 ZK可查看临时节点状态五、总结分布式锁的选择没有绝对的最优解只有最适合场景的解Redis Redlock高性能、易于部署适合可容忍极低概率误放的短事务场景如幂等的定时任务ZooKeeper安全性最高、支持通知和公平性适合零容忍误放的场景如分布式对账、金融扣款混合策略对账类任务用 ZK 锁缓存刷新类任务用 Redis 锁——资源级别的差异化策略在事故复盘后团队将对账任务从 Redis SETNX 切换到了 ZooKeeper 临时顺序节点方案并增加了看门狗机制。切换后从未发生过重复执行。但对于系统中另外 50 多个幂等的缓存刷新任务Redis Redlock 仍然是更合适的选择——在它们上面投入 ZK 的复杂度是过度工程。