引言
有加锁自然就有解锁，本篇则将围绕锁的释放锁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);
}
}
3.两种方式的关键区别：
等待策略：
tryLock：有限时间等待，超时返回false
lock：无限等待直到获取到锁
返回值：
tryLock：返回boolean，表示是否获取成功
lock：无返回值，要么获取成功，要么一直等待
中断处理：
tryLock：支持中断
lock：默认不响应中断，但可以通过lockInterruptibly方法支持中断
小结
关于可重入锁的相关源码刨析就告一段落了，在接下来的篇章中我们将继续分析不同类型锁的实现。
