redis · 2022-03-27

redis 分布式锁 RedissonLock

版本

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>2.9.0</version>
</dependency>

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.4.3</version>
</dependency>

1.加锁原理

加锁其实是通过一段 lua 脚本实现的

加锁：主要是判断锁是否存在，不存在就设置过期时间（首次加锁）；如果锁已经存在了，那对比一下线程，线程一致就证明可以重入 （并刷新过期时间），锁在了但是不是当前线程，证明别人还没释放，那就把剩余时间返回，加锁失败。

redis 存放的是用户指定的 key，key 的 类型是 hash，hash 的 field 是 (UUID : ThreadId)，value 是 1，过期时间默认 30s
相当于执行：

hset redlock 0654df0e-25c4-4e08-a586-cf46d40fd62b:35 1
pexpire redlock 30000

redis 中查看:

127.0.0.1:6379> keys *
1) "redlock"
127.0.0.1:6379> exists "redlock"
(integer) 1
127.0.0.1:6379> type "redlock"
hash
127.0.0.1:6379> hgetall "redlock"
1) "0654df0e-25c4-4e08-a586-cf46d40fd62b:35"
2) "1"
127.0.0.1:6379> pttl "redlock"
(integer) 30000

源码：

// RedissonLock.java
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              "if (redis.call('exists', KEYS[1]) == 0) then " +
                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                  "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "return redis.call('pttl', KEYS[1]);",
                Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

解释如下:

-- ============================================================
-- 参数说明（调用前必看）：
-- KEYS[1]   = 锁的名称，例如 "my_lock"
-- ARGV[1]   = 锁的租约时间（毫秒），例如 30000（30秒）
-- ARGV[2]   = 当前线程唯一标识，例如 "uuid:threadId"
-- ============================================================

-- 第 1 行：条件判断（检查锁是否可获取）
-- 含义：锁不存在（EXISTS 返回 0） 或者 当前线程已持有该锁（HEXISTS 返回 1）
-- 等效 CLI 单步命令：EXISTS my_lock  或  HEXISTS my_lock "uuid:threadId"
if ((redis.call('exists', KEYS[1]) == 0) or (redis.call('hexists', KEYS[1], ARGV[2]) == 1)) then

    -- 第 2 行：可重入计数器 +1
    -- 含义：首次加锁设为 1，重入时累加（例如 1→2，表示重入第 2 次）
    -- 等效 CLI 单步命令：HINCRBY my_lock "uuid:threadId" 1
    redis.call('hincrby', KEYS[1], ARGV[2], 1);

    -- 第 3 行：重置锁的过期时间
    -- 含义：将锁的存活时间刷新为完整的租约时间（防止业务执行过长导致锁提前释放）
    -- 等效 CLI 单步命令：PEXPIRE my_lock 30000
    redis.call('pexpire', KEYS[1], ARGV[1]);

    -- 第 4 行：返回成功标志
    -- 含义：返回 nil 表示加锁或重入成功（客户端收到 nil 即认为获取锁成功）
    -- 等效 CLI 单步命令：（Lua 返回空，无对应 Redis 命令）
    return nil;
end;

-- 第 5 行（条件不满足时执行）：返回锁的剩余存活时间
-- 含义：锁已被其他线程占用，返回剩余毫秒数（供客户端自旋等待）
-- 等效 CLI 单步命令：PTTL my_lock
return redis.call('pttl', KEYS[1]);

2.锁互斥机制

流程：

  1. 尝试获取锁，返回 null 则说明加锁成功，返回一个数值，则说明已经存在该锁，ttl 为锁的剩余存活时间。
  2. 如果此时客户端 2 进程获取锁失败，那么使用客户端 2 的线程 id（其实本质上就是进程 id）通过 Redis 的 channel 订阅锁释放的事件。如果等待的过程中一直未等到锁的释放事件通知，当超过最大等待时间则获取锁失败，返回 false。如果等到了锁的释放事件的通知，则开始进入一个不断重试获取锁的循环。
  3. 循环中每次都先试着获取锁，并得到已存在的锁的剩余存活时间。如果在重试中拿到了锁，则直接返回。如果锁当前还是被占用的，那么等待释放锁的消息，具体实现使用了 JDK 的信号量 Semaphore 来阻塞线程，当锁释放并发布释放锁的消息后，信号量的 release() 方法会被调用，此时被信号量阻塞的等待队列中的一个线程就可以继续尝试获取锁了。

注意：分布式锁的一个关键点，当锁正在被占用时，等待获取锁的进程并不是通过一个 while(true) 死循环去获取锁，而是利用了 Redis 的发布订阅机制,通过 await 方法阻塞等待锁的进程，有效的解决了无效的锁申请浪费资源的问题。

// RedissonLock.java
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long time = unit.toMillis(waitTime);
    long current = System.currentTimeMillis();
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return true;
    }

    time -= System.currentTimeMillis() - current;
    if (time <= 0) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }

    current = System.currentTimeMillis();
    CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    try {
        subscribeFuture.get(time, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        ...
        acquireFailed(waitTime, unit, threadId);
        return false;
    } catch (ExecutionException e) {
        ...
        acquireFailed(waitTime, unit, threadId);
        return false;
    }

    try {
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }

        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                return true;
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }

            // waiting for message
            currentTime = System.currentTimeMillis();
            if (ttl >= 0 && ttl < time) {
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        }
    } finally {
        unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
    }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
}

3.锁的续期机制

Redisson 提供了一个续期机制， 只要线程 一旦加锁成功，就会启动一个 Watch Dog

Watch Dog 机制其实就是一个后台定时任务线程，获取锁成功之后，然后每隔 10 秒 （internalLockLeaseTime / 3） 检查一下，如果线程还持有锁 key（判断线程是否还持有 key，其实根据线程 id 去 Redis 中查，如果存在就会延长 key 的时间），那么就会不断的延长锁 key 的生存时间。

注意：如果服务宕机了，Watch Dog 机制线程也就没有了，此时就不会延长 key 的过期时间，到了 30s 之后就会自动过期了，其他线程就可以获取到锁。

Watch Dog（看门狗）机制的存在，本质上是为了解决分布式锁中一个经典的两难困境（Trade-off）：
锁的过期时间（TTL）设长了，万一服务宕机，锁迟迟不释放，造成死锁；设短了，业务还没执行完，锁就被自动释放，导致并发安全问题。

如果业务代码写了个 while(true) 死循环，且永远不调用 unlock：Watch Dog 每隔 10 秒检查一次，发现“线程 ID 还在”，于是无限续期。锁永远不会自动释放（直到进程挂掉）。这其实是符合设计的，因为 Watch Dog 的宗旨就是“只要你还活着（没调用 unlock），我就保证锁不失效”，它不会去猜测你的业务逻辑是否健康。

// RedissonLock.java
private void scheduleExpirationRenewal(final long threadId) {
    if (expirationRenewalMap.containsKey(getEntryName())) {
        return;
    }

    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {

            RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                    "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return 1; " +
                    "end; " +
                    "return 0;",
                      Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

            future.addListener(new FutureListener<Boolean>() {
                @Override
                public void operationComplete(Future<Boolean> future) throws Exception {
                    expirationRenewalMap.remove(getEntryName());
                    if (!future.isSuccess()) {
                        log.error("Can't update lock " + getName() + " expiration", future.cause());
                        return;
                    }

                    if (future.getNow()) {
                        // reschedule itself
                        scheduleExpirationRenewal(threadId);
                    }
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

    if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
        task.cancel();
    }
}

4.可重入加锁机制

申请锁时，记录锁数量，并续期

相当于执行的是：

hexists redlock 0654df0e-25c4-4e08-a586-cf46d40fd62b:35
hincrby redlock 0654df0e-25c4-4e08-a586-cf46d40fd62b:35 1
pexpire redlock 3000

源码：

// RedissonLock.java
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              "if (redis.call('exists', KEYS[1]) == 0) then " +
                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                  "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "return redis.call('pttl', KEYS[1]);",
                Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

5.释放锁

  1. 删除锁 （查看当前锁重入次数，并减1，如果重入次数为0，则删除锁）
  2. 广播释放锁的消息，通知阻塞等待的进程（向通道名为 redisson_lock__channel publish 一条 unlockMessage 信息，例如通道redisson_lock__channel:{redlock}），重入次数为0，才广播释放锁消息，
  3. 取消 Watch Dog 机制，即将 RedissonLock.expirationRenewalMap 里面的线程 id 删除，并且取消定时任务
// RedissonLock.java
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('exists', KEYS[1]) == 0) then " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; " +
            "end;" +
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                "return nil;" +
            "end; " +
            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
            "if (counter > 0) then " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 0; " +
            "else " +
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; "+
            "end; " +
            "return nil;",
            Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));

}

void cancelExpirationRenewal() {
    Timeout task = expirationRenewalMap.remove(getEntryName());
    if (task != null) {
        task.cancel();
    }
}

解释如下:

-- ============================================================
-- 参数说明（调用前必看）：
-- KEYS[1]   = 锁的名称，例如 "my_lock"
-- KEYS[2]   = 频道名称，例如 "redisson_lock__channel:{my_lock}"
-- ARGV[1]   = 解锁消息（固定值，例如 0）
-- ARGV[2]   = 内部锁租约时间（毫秒），例如 30000（Watch Dog 默认 30 秒）
-- ARGV[3]   = 当前线程唯一标识，例如 "uuid:threadId"
-- ============================================================

-- 第 1 段：锁已经不存在（可能已过期自动释放）
-- 含义：检查锁 Key 是否存在，若不存在（返回 0），说明锁已被 Redis 自动删除
-- 等效 CLI：EXISTS my_lock
if (redis.call('exists', KEYS[1]) == 0) then
    -- 虽然锁不存在，但仍需发布解锁消息，唤醒那些正在等待的线程
    -- 等效 CLI：PUBLISH redisson_lock__channel:{my_lock} 0
    redis.call('publish', KEYS[2], ARGV[1]);
    -- 返回 1 表示解锁操作已完成（虽然是“空解锁”）
    return 1;
end;

-- 第 2 段：锁存在，但当前线程不是持有者（非法解锁）
-- 含义：检查当前线程 ID 是否是该锁的持有者，若不是（返回 0），说明可能误释放他人锁
-- 等效 CLI：HEXISTS my_lock "uuid:threadId"
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
    -- 返回 nil，表示解锁失败（调用方会抛出异常，如 IllegalMonitorStateException）
    return nil;
end;

-- 第 3 段：当前线程是锁持有者 → 可重入计数器 -1
-- 含义：对当前线程 ID 对应的计数器执行减 1 操作（释放一次锁）
-- 等效 CLI：HINCRBY my_lock "uuid:threadId" -1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);

-- 第 4 段：判断计数器是否仍 > 0（还有重入）
-- 含义：counter > 0 说明当前线程还持有该锁（重入未归零）
if (counter > 0) then
    -- 刷新锁的过期时间为完整租约时间（因为是重入，仍需保持锁活跃）
    -- 等效 CLI：PEXPIRE my_lock 30000
    redis.call('pexpire', KEYS[1], ARGV[2]);
    -- 返回 0 表示解锁成功，但锁仍被持有（未完全释放）
    return 0;
else
    -- 第 5 段：计数器归零 → 完全释放锁
    -- 含义：删除锁 Key，并发布解锁消息通知等待的线程
    -- 等效 CLI：DEL my_lock
    redis.call('del', KEYS[1]);
    -- 等效 CLI：PUBLISH redisson_lock__channel:{my_lock} 0
    redis.call('publish', KEYS[2], ARGV[1]);
    -- 返回 1 表示锁已被完全释放
    return 1;
end;

-- 兜底返回（理论上不会执行到这里）
return nil;

栗子

RLock 是 RedissonLock

private static Integer inventory = 1000;

private static final int NUM = 1000;

@Test
public void testRedissonLock() {
    Config config = new Config();
    config.useSingleServer()
            .setAddress("redis://127.0.0.1:6379")
            .setDatabase(0);

    RedissonClient redissonClient =  Redisson.create(config);
    RLock lock = redissonClient.getLock("redlock");

    ThreadPoolExecutor threadPoolExecutor =
            new ThreadPoolExecutor(
                    inventory,
                    inventory,
                    10L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>());

    CyclicBarrier cyclicBarrier = new CyclicBarrier(100);
    long start = System.currentTimeMillis();

    for (int i = 0; i < NUM; i++) {
        threadPoolExecutor.execute(() -> {
            try {
                cyclicBarrier.await();
                lock.lock();
                inventory--;
                System.out.println(inventory);
                lock.unlock();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    while (threadPoolExecutor.getActiveCount() > 0) {
        Thread.yield();
    }

    long end = System.currentTimeMillis();
    System.out.println(String.format("线程执行数:%s 总耗时:%s 库存书:%s",
            NUM,
            (end - start),
            inventory));

    threadPoolExecutor.shutdown();
}

zookeeper 锁 与 redis 锁的比较

对比项 Zookeeper 锁 Redis 锁（Redisson）
CAP 模型 CP（强一致性） AP（高可用，最终一致性）
死锁保障 天然安全：临时节点，客户端会话断开 ZK 自动删除锁，不需要主动续期 依赖过期时间；长任务必须开启看门狗自动续期，看门狗故障仍有死锁风险
锁等待机制 公平锁天然支持：有序节点 + 监听前驱节点，唤醒有序，无惊群效应 原生非公平；Redisson 可实现公平锁，但靠轮询 / 订阅，实现复杂，存在惊群问题
性能 较低。ZK 写操作需要过半节点确认，网络交互多，高并发吞吐量一般 极高。Redis 内存操作，单机能支撑大量锁竞争，适合高吞吐场景
故障风险 主从切换期间：会话有效则临时节点保留；数据一致性强，极少出现锁丢失 主从异步复制！主节点拿到锁后宕机，锁还没同步到从节点，新主节点会导致锁失效（安全漏洞），Redlock 试图解决但有争议
可重入 原生可以自己实现，原生 API 不直接提供 Redisson原生支持可重入锁，开箱即用
阻塞等待 Watcher 事件通知，等待时无空轮询，资源友好 基础实现大量自旋轮询消耗 CPU； Redisson 用 PubSub 优化，但仍不如 ZK 优雅
集群部署成本 较重，至少 3 节点 ZK 集群，运维复杂 轻量，Redis 部署简单，云服务成熟
适用场景 对锁安全性要求极高，并发量中等 高并发、高性能优先，能容忍极小概率锁异常

标签：

您可能还喜欢...

1 条回复

  1. 匿名说道：
    2022-11-03 20:56

    非常好的博客!感谢