分布式锁实现实战：Redis RedLock算法与ZooKeeper方案对比与选型

小编 • • 后端开发

分布式锁是微服务架构中解决并发资源竞争的核心机制。在库存扣减、订单创建、定时任务去重等高并发设计场景中，单机锁无法满足跨进程跨节点的互斥需求。Redis和ZooKeeper是两种主流的分布式锁实现方案，前者追求高性能和低延迟，后者强调一致性和可靠性。理解两种方案的原理和适用边界，是后端开发中服务治理能力的基本要求。

Redis单节点分布式锁实现与SETNX原理解析

Redis分布式锁最基础的实现是SET key value NX EX命令。NX保证只有一个客户端能设置成功，EX设置过期时间防止持锁客户端宕机后死锁。解锁操作必须使用Lua脚本保证原子性，避免误删其他客户端的锁。

import redis
import uuid
import time

class RedisDistributedLock:
    def __init__(self, redis_client, lock_key, expire_time=30):
        self.redis = redis_client
        self.lock_key = lock_key
        self.expire_time = expire_time
        self.lock_value = str(uuid.uuid4())  # 唯一标识当前持有者

    def acquire(self, retry_count=3, retry_delay=0.1):
        '''获取锁'''
        for i in range(retry_count):
            # SET key value NX EX（原子操作）
            result = self.redis.set(
                self.lock_key,
                self.lock_value,
                nx=True,
                ex=self.expire_time
            )
            if result:
                return True
            time.sleep(retry_delay)
        return False

    def release(self):
        '''释放锁（Lua脚本保证原子性）'''
        # 只有锁的持有者才能释放，避免误删
        lua_script = '''
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        '''
        result = self.redis.eval(lua_script, 1, self.lock_key, self.lock_value)
        return result == 1

    def renew(self):
        '''续期锁（防止业务执行时间超过锁过期时间）'''
        lua_script = '''
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("expire", KEYS[1], ARGV[2])
        else
            return 0
        end
        '''
        result = self.redis.eval(
            lua_script, 1, self.lock_key,
            self.lock_value, self.expire_time
        )
        return result == 1

# 使用示例
r = redis.Redis(host='localhost', port=6379, db=0)
lock = RedisDistributedLock(r, 'inventory_lock:product_123', expire_time=10)

if lock.acquire():
    try:
        # 执行业务逻辑
        deduct_inventory('product_123', 1)
        # 如果业务执行时间不确定，启动看门狗续期
        # watchdog = Thread(target=auto_renew, args=(lock,))
        # watchdog.daemon = True
        # watchdog.start()
    finally:
        lock.release()
else:
    print('获取锁失败，请稍后重试')

锁续期（看门狗机制）是生产环境中必须考虑的问题。如果业务执行时间超过锁的过期时间，锁会被自动释放，此时其他客户端可能获取到锁，导致互斥失效。Redisson的看门狗默认每10秒续期一次（锁过期时间30秒），如果续期失败会停止续期并让锁自然过期。

RedLock算法原理与多节点容错实现

单节点Redis锁存在主从切换时锁丢失的风险：主节点加锁后尚未同步到从节点就宕机，哨兵将从节点提升为主节点，新主节点上没有锁记录，其他客户端可以重复加锁。RedLock算法通过在多个独立的Redis实例上同时加锁来解决这个问题。

import redis
import time
import uuid

class RedLock:
    def __init__(self, redis_nodes, lock_key, expire_time=10):
        '''
        redis_nodes: Redis实例列表 [(host, port), ...]
        至少5个独立实例才能保证容错性
        '''
        self.clients = [
            redis.Redis(host=h, port=p, db=0)
            for h, p in redis_nodes
        ]
        self.lock_key = lock_key
        self.expire_time = expire_time  # 毫秒
        self.lock_value = str(uuid.uuid4())
        self.quorum = len(self.clients) // 2 + 1  # 多数派
        self.retry_count = 3
        self.retry_delay = 200  # 毫秒

    def acquire_instance(self, client):
        '''在单个实例上加锁'''
        try:
            return client.set(
                self.lock_key, self.lock_value,
                nx=True, px=self.expire_time
            )
        except Exception:
            return False

    def release_instance(self, client):
        '''在单个实例上释放锁'''
        lua_script = '''
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        '''
        try:
            return client.eval(lua_script, 1, self.lock_key, self.lock_value)
        except Exception:
            return 0

    def acquire(self):
        '''RedLock获取锁'''
        for attempt in range(self.retry_count):
            success_count = 0
            start_time = time.time()

            # 在所有节点上尝试加锁
            for client in self.clients:
                if self.acquire_instance(client):
                    success_count += 1

            # 计算获取锁花费的时间
            elapsed = int((time.time() - start_time) * 1000)

            # 多数派成功且剩余时间充足
            if success_count >= self.quorum and elapsed < self.expire_time:
                # 更新锁的实际有效时间
                self.expire_time -= elapsed
                return True
            else:
                # 加锁失败，释放已获取的锁
                for client in self.clients:
                    self.release_instance(client)

                time.sleep(self.retry_delay / 1000)

        return False

    def release(self):
        '''释放所有节点上的锁'''
        for client in self.clients:
            self.release_instance(client)

# 使用示例
nodes = [
    ('redis1.example.com', 6379),
    ('redis2.example.com', 6379),
    ('redis3.example.com', 6379),
    ('redis4.example.com', 6379),
    ('redis5.example.com', 6379),
]

lock = RedLock(nodes, 'order_lock:user_456', expire_time=10000)
if lock.acquire():
    try:
        process_order('user_456')
    finally:
        lock.release()

RedLock算法的有效性依赖一个前提：各个Redis实例之间是独立的，不存在主从复制关系。如果使用云服务商提供的Redis集群，底层可能共享物理机或网络，在极端情况下可能同时故障。对于绝大多数业务场景，单节点Redis锁配合合理的过期时间已经足够，RedLock适用于对一致性要求极高的金融场景。

ZooKeeper临时顺序节点分布式锁

ZooKeeper通过临时顺序节点实现分布式锁，天然具备锁的可重入性和公平性。临时节点在客户端会话断开时自动删除，解决了持锁者宕机死锁的问题。顺序节点保证了锁的公平获取顺序，先请求的客户端先获得锁。

from kazoo.client import KazooClient
from kazoo.exceptions import NodeExistsError
import time

class ZooKeeperLock:
    def __init__(self, hosts, lock_path, timeout=10):
        self.zk = KazooClient(hosts=hosts, timeout=timeout)
        self.lock_path = lock_path
        self.lock_node = None
        self.locked = False

    def connect(self):
        self.zk.start()
        # 确保锁路径存在
        self.zk.ensure_path(self.lock_path)

    def acquire(self, timeout=None):
        '''获取锁（公平锁）'''
        # 创建临时顺序节点
        self.lock_node = self.zk.create(
            f'{self.lock_path}/lock_',
            ephemeral=True,
            sequence=True
        )
        node_name = self.lock_node.split('/')[-1]

        while True:
            # 获取所有子节点并排序
            children = self.zk.get_children(self.lock_path)
            children.sort()

            # 如果自己是最小节点，获取到锁
            if node_name == children[0]:
                self.locked = True
                return True

            # 否则监听前一个节点
            idx = children.index(node_name)
            prev_node = f'{self.lock_path}/{children[idx - 1]}'

            # 检查前一个节点是否存在
            if self.zk.exists(prev_node, watch=self._watch_handler):
                # 前一个节点存在，等待其删除
                event = self.zk.handler.event_object()
                self._watch_event = event
                event.wait(timeout=timeout)
                if not event.is_set():
                    # 超时
                    self.zk.delete(self.lock_node)
                    return False
            # 前一个节点已删除，重新检查

    def _watch_handler(self, event):
        '''节点删除回调'''
        if hasattr(self, '_watch_event'):
            self._watch_event.set()

    def release(self):
        '''释放锁'''
        if self.locked and self.lock_node:
            self.zk.delete(self.lock_node)
            self.locked = False

    def disconnect(self):
        '''关闭连接'''
        if self.locked:
            self.release()
        self.zk.stop()

# 使用示例
zk_lock = ZooKeeperLock('zk1:2181,zk2:2181,zk3:2181', '/locks/order')
zk_lock.connect()

if zk_lock.acquire(timeout=30):
    try:
        process_order()
    finally:
        zk_lock.release()
zk_lock.disconnect()

# 使用Curator Framework（Java生态推荐）
# InterProcessMutex lock = new InterProcessMutex(curatorClient, "/locks/order");
# if (lock.acquire(30, TimeUnit.SECONDS)) {
#     try {
#         processOrder();
#     } finally {
#         lock.release();
#     }
# }

ZooKeeper锁的优势在于其一致性保证：ZAB协议确保所有事务按顺序持久化，客户端会话超时后临时节点自动清理。劣势是性能低于Redis：每次加锁需要创建节点和设置watch，网络往返开销更大。在锁竞争激烈的场景下，大量watch事件可能造成羊群效应，通过顺序节点监听前一个节点的方式可以缓解。

三种方案对比与生产环境选型建议

Redis单节点锁、RedLock和ZooKeeper锁各有优劣，选型需要根据业务场景的CAP要求决定。

# 方案对比表
'''
| 维度        | Redis单节点锁 | RedLock      | ZooKeeper锁  |
|------------|--------------|-------------|-------------|
| 性能(TPS)  | 10万+        | 1万+        | 1千-5千     |
| 延迟       | ~1ms         | ~5ms        | ~10-50ms    |
| 一致性     | AP（最终一致）| CP+AP混合   | CP（强一致） |
| 可用性     | 高           | 中高        | 中          |
| 复杂度     | 低           | 中          | 中高        |
| 死锁风险   | 依赖过期时间  | 依赖过期时间 | 无（会话超时）|
| 公平性     | 非公平       | 非公平      | 公平        |
| 可重入性   | 需自行实现    | 需自行实现  | 原生支持    |
'''

# 工厂模式：根据场景选择锁实现
class LockFactory:
    @staticmethod
    def create(lock_type, config):
        if lock_type == 'redis':
            return RedisDistributedLock(
                redis.Redis(host=config['host'], port=config['port']),
                config['key'],
                expire_time=config.get('ttl', 30)
            )
        elif lock_type == 'redlock':
            return RedLock(
                config['nodes'],
                config['key'],
                expire_time=config.get('ttl', 10000)
            )
        elif lock_type == 'zookeeper':
            lock = ZooKeeperLock(
                config['hosts'],
                config['path'],
                timeout=config.get('timeout', 10)
            )
            lock.connect()
            return lock

# 选型建议：
# 1. 电商库存扣减、秒杀 -> Redis单节点锁（高并发，允许极小概率的不一致）
# 2. 金融转账、支付 -> ZooKeeper锁（强一致，可接受性能损失）
# 3. 定时任务去重 -> Redis单节点锁（低频，简单可靠）
# 4. 跨数据中心分布式锁 -> RedLock（多节点容错，但需独立部署Redis）

消息中间件配合分布式锁使用可以进一步提升系统的可靠性。将锁获取和业务执行封装为消息消费过程，利用消息队列的重试和死信机制处理锁获取失败的请求。这种模式在业务中台建设中广泛应用，将分布式锁的复杂性封装在中间件层，业务代码保持简洁。

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

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

相关推荐