其他数据库 2023-02-22

Redisson 加锁解锁的实现

分布式锁使用

对于 redisson 分布式锁的使用很简单：

1、调用 getLock 函数获取锁操作对象；
2、调用 tryLock 函数进行加锁；
3、调用 unlock 函数进行解锁；

注意 unlock 操作需要放到 finally 代码段中，保证锁可以被释放。

private void sumLock() {
lock = redissonClient.getLock(“sum-lock”);

boolean b = lock.tryLock();
if (!b) {
log.info(“获取不到锁”);
return;
}

try {
for (int j = 0; j < 20000; j++) {
++sum;
}
} finally {
lock.unlock();
}

}

getLock

getLock 实例化 RedissonLock，相当于 Lock lock = new ReentrantLock() 操作；

public RLock getLock(String name) {
// 实例化 RedissonLock，参数为指令执行器和锁名称
return new RedissonLock(this.connectionManager.getCommandExecutor(), name);
}

public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
super(commandExecutor, name);
// 命令执行器，用于执行lua脚本
this.commandExecutor = commandExecutor;
// 连接管理器的ID
this.id = commandExecutor.getConnectionManager().getId();
// 锁续期时间（看门狗），锁默认续期时间是 30s。
this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
this.entryName = this.id + “:” + name;
this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
}

tryLock

@Override
public boolean tryLock() {
return get(tryLockAsync());
}

@Override
public RFuture<Boolean> tryLockAsync() {
return tryLockAsync(Thread.currentThread().getId());
}

@Override
public RFuture<Boolean> tryLockAsync(long threadId) {
return tryAcquireOnceAsync(-1, -1, null, threadId);
}

这里是一系列的调用，可以直接跳过，直接进入到 tryAcquireOnceAsync 函数，看看 tryAcquireOnceAsync 函数的处理逻辑。

private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
// 由于我们调用tryLocak没有传递任何参数，leaseTime默认为-1，不走判断
if (leaseTime != -1) {
return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
}

// 调用获取锁枷锁的主要逻辑在这里
RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);

ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
// 如果发生异常那么直接放回了
if (e != null) {
return;
}

// 锁续期
if (ttlRemaining) {
scheduleExpirationRenewal(threadId);
}
});
// 返回结果
return ttlRemainingFuture;
}

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
// 将时间转化为毫秒
internalLockLeaseTime = unit.toMillis(leaseTime);
// 执行脚本
return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
“if (redis.call(‘exists’, KEYS[1]) == 0) then ” +
“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 ” +
“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()), internalLockLeaseTime, getLockName(threadId));
}

Redisson 中存储锁的数据类型结构采用的的是 hash，Key 为锁名称，VALUE的属性是 Redisson 客户端ID和线程ID组合而成的字符串，值是锁的重入次数，采用 hash 计数实现锁的重入性。

该函数主要执行 lua 脚本，脚本的逻辑为：

1、redis.call(‘exists’, KEYS[1]) == 0 用于判断锁是否存在，等于 0 说明不存在，表明此时没有客户端持有锁，此客户端获取锁成功；走步骤 2，否则走步骤 4；
2、设置锁，并且对锁进行 +1 操作，标识获取锁的次数；
3、为锁设置过期时间，成功返回 nil；
4、redis.call(‘hexists’, KEYS[1], ARGV[2]) == 1 判断锁是否本客户端持有，等于1说明是，此时是再次获取锁（重入），走步骤 5，否则走 7；
5、对锁进行 +1 操作，标识获取锁的次数；
6、为锁设置过期时间，成功返回 nil；
7、如果1和4的判断都不满足，那么返回锁的的剩余时间；

unLock

@Override
public void unlock() {
try {
get(unlockAsync(Thread.currentThread().getId()));
} catch (RedisException e) {
if (e.getCause() instanceof IllegalMonitorStateException) {
throw (IllegalMonitorStateException) e.getCause();
} else {
throw e;
}
}
}

@Override
public RFuture<Void> unlockAsync(long threadId) {
RPromise<Void> result = new RedissonPromise<Void>();
// 释放锁的逻辑主要这里
RFuture<Boolean> future = unlockInnerAsync(threadId);

future.onComplete((opStatus, e) -> {
cancelExpirationRenewal(threadId);

if (e != null) {
result.tryFailure(e);
return;
}

if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException(“attempt to unlock lock, not locked by current thread by node id: ”
+ id + ” thread-id: ” + threadId);
result.tryFailure(cause);
return;
}

result.trySuccess(null);
});

return result;
}

这里是一系列的调用，可以直接跳过，直接进入到 unlockInnerAsync 函数，看看 unlockInnerAsync 函数的处理逻辑。

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return 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); ” +
“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.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}

该函数主要执行 lua 脚本，脚本的逻辑为：

1、redis.call(‘hexists’, KEYS[1], ARGV[3]) == 0 用于判断锁是否为当前客户端持有，等于 0 说明不是直接返回 nil，否则说明是，走步骤 2；
2、对锁进行 -1 操作，并且获取其计数 counter；
3、判断 counter > 0，如果大于 0 说明该客户端多次获取锁，对锁进行续期并且返回 0，因为此时业务还没有执行完毕，否则走步骤 4；
4、如果count 小于等于 0 则删除锁，发送释放锁的消息，返回 1;
5、如果以上逻辑都不满足，那么直接返回nil；

总结

redisson 加锁解锁：

1、redisson 使用 lua 脚本保证命令执行的原子性；
2、redisson 使用 redis 的 hash 数据结构类型来存储锁信息，使用锁名称作为 hash 名称，使用“客户端ID:线程ID”作为键，使用重入次数作为值；
3、每次获取锁，会对值进行 +1 操作，并且设置过期时间；
4、每次释放锁，会对值进行 -1 操作，如果没有减少为 0，则继续设置锁的超时时间，否则删除锁，并且发送释放锁的消息。

到此这篇关于Redisson 加锁解锁的实现的文章就介绍到这了,更多相关Redisson 加锁解锁内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！

NoSQL Redis 技术文档数据库

数据运维技术 » Redisson 加锁解锁的实现

分享到：

分布式锁使用

getLock

tryLock

unLock

总结

相关推荐