Redis分布式锁实现方案与Redlock算法一致性分析实战

小编 • • 后端开发

分布式锁是高并发设计中的基础组件，用于在分布式系统中协调多个节点对共享资源的互斥访问。在微服务架构中，秒杀库存扣减、定时任务防重复执行、订单状态流转等场景都依赖分布式锁保证操作原子性。Redis作为高性能内存数据库，是实现分布式锁的主流方案，但其单节点方案在故障转移时存在锁失效风险，Redlock算法通过多节点投票机制提升锁的可靠性。

分布式锁应用场景与实现挑战

分布式锁需要满足三个核心属性：互斥性（同一时刻只有一个客户端持有锁）、可重入性（持有锁的客户端可以再次获取锁）、容错性（部分节点故障不影响锁服务）。实现挑战主要在于锁的获取与释放必须保证原子性，以及网络分区或节点故障时的锁状态一致性。

典型应用场景：电商秒杀活动中，多个请求同时扣减库存，需要分布式锁保证库存不为负数。定时任务在多实例部署时，只需一个实例执行，通过分布式锁实现选主。分布式事务中防止重复提交，使用锁作为前置校验。

Redis单节点分布式锁实现

最基础的Redis分布式锁使用SET命令的NX（不存在才设置）和PX（过期时间）选项：

import redis
import uuid

class RedisLock:
    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.1):
        for i in range(retry_count):
            result = self.redis.set(
                self.lock_key, self.lock_value,
                nx=True, ex=self.expire_seconds
            )
            if result:
                return True
            time.sleep(retry_delay)
        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
        '''
        result = self.redis.eval(lua_script, 1, self.lock_key, self.lock_value)
        return result == 1

释放锁使用Lua脚本而非先GET再DEL，因为GET和DEL之间存在时间窗口：锁可能在此期间过期并被其他客户端获取，DEL会误删新锁。Lua脚本在Redis中原子执行，消除这个竞态条件。

可重入锁实现。持有锁的客户端需要多次获取同一把锁时，使用Hash结构记录重入次数：

class ReentrantRedisLock:
    def __init__(self, redis_client, lock_key, client_id, expire_seconds=30):
        self.redis = redis_client
        self.lock_key = lock_key
        self.client_id = client_id
        self.expire_seconds = expire_seconds

    def acquire(self):
        lua_script = '''
        if redis.call("exists", KEYS[1]) == 0 then
            redis.call("hset", KEYS[1], ARGV[1], 1)
            redis.call("expire", KEYS[1], ARGV[2])
            return 1
        elseif redis.call("hexists", KEYS[1], ARGV[1]) == 1 then
            redis.call("hincrby", KEYS[1], ARGV[1], 1)
            redis.call("expire", KEYS[1], ARGV[2])
            return 1
        else
            return 0
        end
        '''
        result = self.redis.eval(lua_script, 1, self.lock_key, self.client_id, self.expire_seconds)
        return result == 1

    def release(self):
        lua_script = '''
        if redis.call("hexists", KEYS[1], ARGV[1]) == 0 then
            return nil
        end
        local count = redis.call("hincrby", KEYS[1], ARGV[1], -1)
        if count <= 0 then
            redis.call("del", KEYS[1])
        else
            redis.call("expire", KEYS[1], ARGV[2])
        end
        return count
        '''
        result = self.redis.eval(lua_script, 1, self.lock_key, self.client_id, self.expire_seconds)
        return result

SETNX锁的典型问题与解决方案

SET NX PX方案存在三个已知问题：

问题一：锁过期但业务未执行完。如果持锁客户端因GC暂停或网络延迟导致业务执行时间超过锁过期时间，锁会自动释放，其他客户端获取锁后可能导致数据不一致。解决方案是Watchdog自动续期机制：持锁客户端启动后台线程定期延长锁过期时间。

import threading
import time

class RedisLockWithWatchdog:
    def __init__(self, redis_client, lock_key, expire_seconds=30):
        self.redis = redis_client
        self.lock_key = lock_key
        self.lock_value = str(uuid.uuid4())
        self.expire_seconds = expire_seconds
        self._running = False

    def acquire(self):
        result = self.redis.set(self.lock_key, self.lock_value, nx=True, ex=self.expire_seconds)
        if result:
            self._running = True
            thread = threading.Thread(target=self._watchdog, daemon=True)
            thread.start()
            return True
        return False

    def _watchdog(self):
        renew_interval = self.expire_seconds / 3
        lua_renew = '''
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("expire", KEYS[1], ARGV[2])
        else
            return 0
        end
        '''
        while self._running:
            time.sleep(renew_interval)
            try:
                result = self.redis.eval(lua_renew, 1, self.lock_key, self.lock_value, self.expire_seconds)
                if result == 0:
                    break
            except Exception:
                break

    def release(self):
        self._running = False
        lua_script = '''
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        '''
        self.redis.eval(lua_script, 1, self.lock_key, self.lock_value)

Redisson框架内置了Watchdog机制，默认每10秒续期一次（锁默认30秒过期）。使用Redisson时不需要手动实现续期逻辑。

问题二：主从切换导致锁丢失。Redis主从复制是异步的，主节点写入锁后未同步到从节点时主节点宕机，哨兵将从节点提升为主节点，新主节点上没有锁记录，其他客户端可以获取同一把锁。这是单节点Redis分布式锁的根本性缺陷，Redlock算法专门解决此问题。

问题三：客户端长时间阻塞。客户端获取锁后如果发生长时间GC或网络中断，锁已过期但客户端不自知，继续操作共享资源。解决方案是在操作前检查锁是否仍持有，但无法完全避免竞态。

Redlock算法原理与多节点一致性保证

Redlock算法由Redis作者Antirez提出，通过在多个独立Redis节点上同时获取锁来提升容错性。核心思想是：在N个独立Redis节点上获取锁，当成功在多数节点（N/2+1）上获取锁时，锁获取成功。

import redis
import time
import uuid
import random

class Redlock:
    def __init__(self, redis_nodes, lock_key, expire_ms=30000, retry_count=3):
        self.clients = [redis.Redis(**node, socket_timeout=50) for node in redis_nodes]
        self.lock_key = lock_key
        self.expire_ms = expire_ms
        self.retry_count = retry_count
        self.quorum = len(self.clients) // 2 + 1
        self.lock_value = str(uuid.uuid4())

    def acquire(self):
        for attempt in range(self.retry_count):
            success_count = 0
            start_time = time.monotonic()
            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 redis.RedisError:
                    pass
            elapsed_ms = (time.monotonic() - start_time) * 1000
            if success_count >= self.quorum and elapsed_ms < self.expire_ms:
                self.validity_time = self.expire_ms - int(elapsed_ms)
                return True
            else:
                self._release_partial()
                time.sleep(0.05 + random.random() * 0.1)
        return False

    def _release_partial(self):
        lua_script = '''
        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_script, 1, self.lock_key, self.lock_value)
            except redis.RedisError:
                pass

    def release(self):
        self._release_partial()

Redlock的关键设计点：

1. 客户端对每个节点的请求设置短超时（如50ms），避免单个节点网络问题阻塞整体流程。

2. 获取锁后计算消耗时间elapsed_ms，如果接近过期时间则视为获取失败，因为剩余有效期不足以完成业务操作。

3. 获取失败时释放所有已获取的锁，避免残留锁阻塞其他客户端。

4. 重试时加入随机延迟，避免多个客户端同时重试导致脑裂。

分布式锁续约与Watchdog机制

Redlock同样面临锁过期但业务未完成的问题。续约机制在Redlock场景下更复杂，需要对多数节点续约成功才算续约有效：

def renew_redlock(self):
    lua_renew = '''
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("pexpire", KEYS[1], ARGV[2])
    else
        return 0
    end
    '''
    success_count = 0
    for client in self.clients:
        try:
            result = client.eval(lua_renew, 1, self.lock_key, self.lock_value, self.expire_ms)
            if result == 1:
                success_count += 1
        except redis.RedisError:
            pass
    return success_count >= self.quorum

续约需要在锁过期前完成。Watchdog线程按1/3有效期间隔续约，每次续约验证多数节点响应。如果续约失败（如部分节点故障），需要立即终止业务操作并释放锁，因为锁已不再有效。

分布式锁性能测试与选型建议

单节点Redis锁与Redlock的性能对比（1000次获取释放，10并发线程）：

单节点锁：平均获取延迟0.8ms，99分位2.1ms，吞吐量约12000次/秒。

Redlock（5节点）：平均获取延迟3.2ms，99分位8.5ms，吞吐量约3000次/秒。

Redlock延迟增加来自对多个节点的串行请求，且每次需要等待多数响应。对延迟敏感的场景需要权衡一致性与性能。

选型建议：对于大多数业务场景，单节点Redis锁配合Watchdog续期已经足够，主从切换导致锁丢失的概率极低。Redlock适用于金融交易、库存扣减等对一致性要求极高的场景。如果业务对正确性要求绝对严格（如分布式事务协调），建议使用etcd或ZooKeeper的分布式锁，它们基于Raft/ZAB共识算法，一致性保证更强。

使用Redisson简化实现。Redisson是Java生态中成熟的Redis客户端，内置分布式锁实现：

Config config = new Config();
config.useSentinelServers()
    .setMasterName("mymaster")
    .addSentinelAddress("redis://node1:6379")
    .addSentinelAddress("redis://node2:6379")
    .addSentinelAddress("redis://node3:6379");

RedissonClient client = Redisson.create(config);
RLock lock = client.getLock("orderLock:12345");
try {
    if (lock.tryLock(5, 30, TimeUnit.SECONDS)) {
        try {
            processOrder();
        } finally {
            lock.unlock();
        }
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

Redisson的lock.tryLock方法在不指定leaseTime参数时自动启用Watchdog，默认每10秒续期一次，续期到30秒。锁释放时Watchdog自动停止。如果使用Redlock，Redisson提供RLock数组形式的联锁：

RLock lock1 = client1.getLock("lockKey");
RLock lock2 = client2.getLock("lockKey");
RLock lock3 = client3.getLock("lockKey");

RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
try {
    redLock.lock(30, TimeUnit.SECONDS);
    // 业务逻辑
} finally {
    redLock.unlock();
}

RedissonRedLock在多个Redisson客户端实例上获取锁，实现Redlock算法。注意Redisson自4.12.0版本起将RedissonRedLock标记为Deprecated，官方建议使用RLock联锁（multiLock）替代，但Redlock算法逻辑在multiLock中保留，仅API形式变化。

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

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

相关推荐