【📕分布式锁通关指南 06】源码剖析redisson可重入锁之加锁

原创

别惹CC

发布于 2025-03-04 11:24:30

710

文章被收录于专栏：分布式锁通关指南分布式锁通关指南

引言

在上篇中，我们基于spring boot整合redisson实现了分布式锁，接下来我会带领大家花一些时间来学习redisson如何实现各种锁，所以我们需要先从github上下载它的源码，本篇则先从可重入锁的相关实现开始来为大家做讲解。

加锁流程分析

这里我们按照步骤逐步分析Redisson 可重入锁的加锁流程。

1.首先从入口方法开始 (RLock.lock())：

// RLock 接口的默认实现类 RedissonLock
public void lock() {
    try {
        lock(-1, null, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

2.核心加锁逻辑实现：

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    
    // 获取当前线程ID
    long threadId = Thread.currentThread().getId();
    
    // 尝试获取锁
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    
    // 如果ttl为空，表示获取锁成功
    if (ttl == null) {
        return;
    }
    
    // 如果获取锁失败，订阅到对应的redisson锁channel，等待锁释放消息
    RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    if (interruptibly) {
        subscribeFuture.syncUninterruptibly();
    } else {
        subscribeFuture.sync();
    }

    try {
        while (true) {
            // 再次尝试获取锁
            ttl = tryAcquire(leaseTime, unit, threadId);
            // 成功获取锁，直接返回
            if (ttl == null) {
                break;
            }
            
            // 等待锁释放通知
            if (ttl >= 0) {
                try {
                    await(ttl, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                }
            }
        }
    } finally {
        // 取消订阅
        unsubscribe(subscribeFuture, threadId);
    }
}

3.tryAcquire 方法实现（这里包含了可重入的核心逻辑）：

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
    
    // 根据传入的租约时间计算锁的过期时间
    long currentTime = System.currentTimeMillis();
    Long ttl = null;
    
    // 如果指定了租约时间
    if (leaseTime != -1) {
        ttl = tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    } 
    // 使用默认的过期时间
    else {
        ttl = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), 
            TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        // 启动看门狗定时续期
        scheduleExpirationRenewal(threadId);
    }
    
    return ttl;
}

梳理一下整个加锁流程：

1. 入口调用：

用户调用 lock() 方法开始加锁
默认使用无限等待时间，且不响应中断

2. 加锁尝试：

首先获取当前线程 ID
调用 tryAcquire 尝试获取锁
如果获取成功（返回 null），则直接返回
如果获取失败，进入等待流程

3. 锁等待流程：

订阅锁释放的 Channel，等待通知
进入循环，不断尝试获取锁
获取成功则退出循环
获取失败则等待指定时间后继续尝试

加锁Lua脚本分析

继续看tryLockInner方法 - 它是最核心的加锁 Lua 脚本：

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    // 这里执行 Lua 脚本
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
        // 判断锁是否存在
        "if (redis.call('exists', KEYS[1]) == 0) then " +
            // 不存在则创建锁，并设置重入次数为1
            "redis.call('hincrby', 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 " +
            // 是当前线程，则重入次数+1
            "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.singletonList(getName()), // KEYS[1] 锁名称
        unit.toMillis(leaseTime), // ARGV[1] 锁过期时间
        getLockName(threadId)); // ARGV[2] 线程标识
}

它的核心逻辑也很简单：首先检查锁是否存在，如果不存在，则直接加锁，且设置重入次数为1；如果存在，先检查是否是当前线程的锁，如果是，则重入次数+1，如果不是，则返回锁的剩余过期时间。