分布式锁实战：Redis RedLock算法与Zookeeper互斥锁方案对比

小编 • • 后端开发

分布式锁在多节点部署环境中保证同一时刻只有一个实例执行临界区操作。常见的实现方案包括基于Redis的单节点锁和RedLock多节点锁、基于Zookeeper临时节点的锁，以及基于数据库唯一约束的锁。不同方案在一致性保证、性能表现和故障恢复方面各有取舍。

Redis单节点分布式锁实现

Redis分布式锁的核心是SET key value NX PX命令——NX保证互斥（key不存在才设置），PX设置过期时间防止死锁。value使用唯一标识（如UUID+线程ID）用于安全释放。

import redis
import uuid
import time

class RedisDistributedLock:
    def __init__(self, redis_client, lock_key, expire_ms=30000):
        self.redis = redis_client
        self.lock_key = lock_key
        self.expire_ms = expire_ms
        self.lock_value = str(uuid.uuid4())
    
    def acquire(self, retry_count=3, retry_delay=100):
        for i in range(retry_count):
            result = self.redis.set(
                self.lock_key, self.lock_value, nx=True, px=self.expire_ms
            )
            if result:
                return True
            time.sleep(retry_delay / 1000)
        return False
    
    def release(self):
        lua = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        result = self.redis.eval(lua, 1, self.lock_key, self.lock_value)
        return result == 1

释放锁必须使用Lua脚本，而非先GET再DEL。后者存在竞态条件：客户端A判断value匹配后、执行DEL前锁过期，客户端B获取了锁，此时A的DEL会误删B的锁。

锁续期与看门狗机制

固定过期时间的问题在于：业务执行时间超过锁过期时间，锁被自动释放，其他实例获取锁后产生并发冲突。Redisson的看门狗（Watchdog）机制通过后台线程定期续期解决此问题：

// Java Redisson实现 - 自动续期
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6379");
RedissonClient redisson = Redisson.create(config);

RLock lock = redisson.getLock("order:lock:1001");

try {
    // 不指定leaseTime则触发看门狗自动续期
    // 看门狗默认每10秒续期一次，将过期时间重置为30秒
    boolean acquired = lock.tryLock(5, 30, TimeUnit.SECONDS);
    if (acquired) {
        processOrder();
    }
} finally {
    if (lock.isHeldByCurrentThread()) {
        lock.unlock();
    }
}
# Python手动续期实现
import threading

class RedisLockWithWatchdog:
    def __init__(self, redis_client, lock_key, expire_ms=30000):
        self.redis = redis_client
        self.lock_key = lock_key
        self.expire_ms = expire_ms
        self.lock_value = str(uuid.uuid4())
        self._stop_event = threading.Event()
        self._watchdog_thread = None
    
    def acquire(self):
        result = self.redis.set(self.lock_key, self.lock_value, nx=True, px=self.expire_ms)
        if result:
            self._start_watchdog()
            return True
        return False
    
    def _start_watchdog(self):
        interval = self.expire_ms / 3 / 1000
        def watchdog():
            while not self._stop_event.wait(interval):
                lua = """
                if redis.call("get", KEYS[1]) == ARGV[1] then
                    return redis.call("pexpire", KEYS[1], ARGV[2])
                else
                    return 0
                end
                """
                self.redis.eval(lua, 1, self.lock_key, self.lock_value, self.expire_ms)
        self._watchdog_thread = threading.Thread(target=watchdog, daemon=True)
        self._watchdog_thread.start()
    
    def release(self):
        self._stop_event.set()
        lua = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(lua, 1, self.lock_key, self.lock_value)

RedLock多节点锁算法

单节点Redis锁在主从切换时存在丢失风险：主节点加锁成功后宕机，从节点晋升为主节点但未同步到锁数据。RedLock通过在多个独立Redis实例上同时加锁，多数成功才算获取成功：

import redis
import time
import uuid
import random

class RedLock:
    def __init__(self, redis_nodes, lock_key, expire_ms=30000):
        self.clients = [redis.Redis(host=h, port=p) for h, p in redis_nodes]
        self.lock_key = lock_key
        self.expire_ms = expire_ms
        self.lock_value = str(uuid.uuid4())
        self.quorum = len(redis_nodes) // 2 + 1
    
    def acquire(self, retry_count=3, retry_delay=200):
        for attempt in range(retry_count):
            success_count = 0
            start_time = time.time()
            for client in self.clients:
                try:
                    result = client.set(
                        self.lock_key, self.lock_value, nx=True, px=self.expire_ms
                    )
                    if result:
                        success_count += 1
                except Exception:
                    continue
            elapsed_ms = (time.time() - start_time) * 1000
            remaining_ttl = self.expire_ms - elapsed_ms
            if success_count >= self.quorum and remaining_ttl > 0:
                return True
            else:
                self._release_all()
                time.sleep(retry_delay / 1000 + random.random() * 0.1)
        return False
    
    def _release_all(self):
        lua = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        for client in self.clients:
            try:
                client.eval(lua, 1, self.lock_key, self.lock_value)
            except Exception:
                continue
    
    def release(self):
        self._release_all()

Zookeeper分布式锁实现

Zookeeper分布式锁基于临时顺序节点：每个客户端在锁节点下创建顺序临时节点，序号最小的获取锁，其他客户端监听前一个节点的删除事件。这种方案天然公平且能感知客户端宕机（临时节点随Session断开自动删除）。

from kazoo.client import KazooClient
from kazoo.exceptions import NoNodeError
import threading

class ZkDistributedLock:
    def __init__(self, zk_hosts, lock_path, timeout=30):
        self.zk = KazooClient(hosts=zk_hosts)
        self.lock_path = lock_path
        self.timeout = timeout
        self.lock_node = None
        self.lock_name = f"{lock_path}/lock_"
        self._watch_event = None
    
    def connect(self):
        self.zk.start(timeout=self.timeout)
    
    def acquire(self):
        self.zk.ensure_path(self.lock_path)
        self.lock_node = self.zk.create(
            self.lock_name, ephemeral=True, sequence=True
        )
        while True:
            children = self.zk.get_children(self.lock_path)
            children_sorted = sorted(children)
            my_node = self.lock_node.split('/')[-1]
            my_index = children_sorted.index(my_node)
            if my_index == 0:
                return True
            prev_node = f"{self.lock_path}/{children_sorted[my_index - 1]}"
            self._watch_event = threading.Event()
            if self.zk.exists(prev_node, watch=self._on_node_deleted):
                self._watch_event.wait()
    
    def _on_node_deleted(self, event):
        if self._watch_event:
            self._watch_event.set()
    
    def release(self):
        if self.lock_node:
            try:
                self.zk.delete(self.lock_node)
            except NoNodeError:
                pass
            self.lock_node = None
    
    def close(self):
        self.zk.stop()
        self.zk.close()

Redis锁与Zookeeper锁对比选型

两种方案在一致性强度和性能上有明显差异：

对比维度          | Redis单节点锁    | RedLock        | Zookeeper锁
----------------|-----------------|----------------|------------------
一致性保证        | AP（最终一致）   | 自定义（多数）  | CP（强一致）
加锁性能          | ~1ms            | ~5ms（多节点）  | ~10ms（创建节点）
故障恢复          | 依赖过期时间     | 多数存活即可用   | 临时节点自动删除
公平性            | 非公平           | 非公平          | 公平（顺序节点）
客户端宕机处理    | 等待过期         | 等待过期        | Session断开自动释放
部署复杂度        | 低               | 高（5+实例）     | 中（ZK集群）

分布式锁的误用场景与替代方案

分布式锁不是万能的。以下场景应考虑替代方案：

库存扣减：使用Redis Lua脚本原子扣减或数据库UPDATE…WHERE stock > 0，比分布式锁更高效：

-- 数据库原子扣减，无需分布式锁
UPDATE products SET stock = stock - 1 
WHERE id = 1001 AND stock > 0;

-- Redis Lua原子扣减
local stock = redis.call('get', KEYS[1])
if tonumber(stock) >= tonumber(ARGV[1]) then
    redis.call('decrby', KEYS[1], ARGV[1])
    return 1
else
    return 0
end

幂等控制：使用唯一索引或Redis SETNX标记，而非长时间持有锁。分布式锁的持有时间应尽可能短，聚焦在无法通过原子操作解决的复杂临界区场景。

选型上，Java技术栈推荐Redisson（内置看门狗和重入锁实现），Python/Go推荐自行封装Redis锁或使用etcd客户端。Zookeeper方案适合对一致性要求极高且已有ZK基础设施的场景。

原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/fen-bu-shi-suo-shi-zhan-redisredlock-suan-fa-yu-zookeeper/

赞 (0)
小编小编
0
上一篇 8小时前
下一篇 8小时前

相关推荐