【分布式锁通关指南 08】源码剖析redisson可重入锁之释放及阻塞与非阻塞获取

引言

有加锁自然就有解锁，本篇则将围绕锁的释放锁Lua脚本进行深入剖析，另外，还将对阻塞和非阻塞两张方式分别如何获取锁进行比较。

可重入锁之释放锁

这里我们依然是按照步骤来看看释放锁是如何执行的。

1.首先从入口方法开始：

public void unlock() {try {get(unlockAsync(Thread.currentThread().getId()));} catch (RedisException e) {if (e.getCause() instanceof IllegalMonitorStateException) {throw (IllegalMonitorStateException) e.getCause();} else {throw e;}}
}// 异步解锁方法
public RFuture<Void> unlockAsync(long threadId) {// 调用解锁的 Lua 脚本RFuture<Boolean> future = unlockInnerAsync(threadId);return future.thenAccept((opStatus) -> {// 解锁成功，取消看门狗续期cancelExpirationRenewal(threadId);// 如果解锁的不是自己的锁，抛出异常if (!opStatus) {throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "+ id + " thread-id: " + threadId);}});
}

2.核心解锁Lua脚本实现：

protected RFuture<Boolean> unlockInnerAsync(long threadId) {return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,// 检查锁是否存在"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +"return nil;" +"end; " +// 计算当前线程的重入次数"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +// 如果重入次数还大于0，则更新过期时间"if (counter > 0) then " +"redis.call('pexpire', KEYS[1], ARGV[2]); " +"return 0; " +"end; " +// 重入次数为0，删除锁"redis.call('del', KEYS[1]); " +// 发布锁释放的消息"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; ",// 脚本参数Arrays.asList(getName(),                // KEYS[1] 锁名称getChannelName(),         // KEYS[2] 发布订阅的channel名称RedissonLockEntry.UNLOCK_MESSAGE,    // ARGV[1] 解锁消息internalLockLeaseTime,    // ARGV[2] 锁过期时间getLockName(threadId)     // ARGV[3] 线程标识));
}

3.梳理流程

首先进行解锁的前置检查：检查是否存在对应线程的锁，如果不存在，则返回nil。
如果获取锁成功，则：处理重入计数，即将当前线程的重入计数减1；如果重入计数还大于0，表示还有重入，则重新设置过期时间，返回0则表示锁还未完全释放。
完全释放锁，即：当计数器为0，删除整个锁并发布锁释放的消息，通知等待的线程，返回1则表示锁已完全释放。
后续处理，需要：解锁成功后取消看门狗续期和处理异常情况。

可重入锁之阻塞和非阻塞获取锁

redisson提供了两种不同方式获取锁的封装，我们这里比较讲下：

1.非阻塞获取锁 (tryLock)

public boolean tryLock() {return tryLock(-1, -1, TimeUnit.MILLISECONDS);
}// 带超时的非阻塞获取锁
public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {return tryLock(waitTime, -1, unit);
}// 核心实现
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(leaseTime, unit, threadId);// ttl为空表示获取成功if (ttl == null) {return true;}// 如果没有等待时间，直接返回失败if (time <= 0) {return false;}// 计算剩余等待时间time -= System.currentTimeMillis() - current;if (time <= 0) {return false;}// 订阅锁释放通知RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {return false;}try {while (true) {long currentTime = System.currentTimeMillis();ttl = tryAcquire(leaseTime, unit, threadId);// 获取成功if (ttl == null) {return true;}// 超过等待时间，返回失败time -= System.currentTimeMillis() - currentTime;if (time <= 0) {return false;}// 等待锁释放通知currentTime = System.currentTimeMillis();if (ttl >= 0 && ttl < time) {getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);}time -= System.currentTimeMillis() - currentTime;if (time <= 0) {return false;}}} finally {// 取消订阅unsubscribe(subscribeFuture, threadId);}
}

2.阻塞获取锁 (Lock)

public void lock() {try {lock(-1, null, false);} catch (InterruptedException e) {throw new IllegalStateException();}
}// 带超时的阻塞获取锁
public void lock(long leaseTime, TimeUnit unit) {try {lock(leaseTime, unit, false);} catch (InterruptedException e) {throw new IllegalStateException();}
}// 核心实现
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {long threadId = Thread.currentThread().getId();// 第一次尝试获取锁Long ttl = tryAcquire(leaseTime, unit, threadId);if (ttl == null) {return;}// 订阅锁释放通知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 {getEntry(threadId).getLatch().await(ttl, TimeUnit.MILLISECONDS);} catch (InterruptedException e) {if (interruptibly) {throw e;}getEntry(threadId).getLatch().await();}} else {if (interruptibly) {getEntry(threadId).getLatch().await();} else {getEntry(threadId).getLatch().awaitUninterruptibly();}}}} finally {// 取消订阅unsubscribe(subscribeFuture, threadId);}
}