Redisson 分布式锁
简单入门
pom
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>
配置类
package com.hmdp.config;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* redisson的配置类
* @author jjking
* @date 2023-11-06 20:21
*/
@Configuration
public class RedissonConfig {
@Bean
public RedissonClient redissonClient() {
//配置
Config config = new Config();
config.useSingleServer().setAddress("redis://8.140.54.97:6379");
//创建RedissonClient对象
return Redisson.create(config);
}
}
简单的使用
获取锁的方法的参数
看官网
RLock lock = redisson.getLock("myLock");
//默认锁
lock.lock();
//锁的过期时间为10s
lock.lock(10, TimeUnit.SECONDS);
//获取锁失败,重试时间为100s,获取锁成功,锁过期时间为10s
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
try {
...
} finally {
lock.unlock();
}
}
可重入锁原理
redissson的可重入原理和Reentranlock很像,都是用计数器来实现的
因为这里既要锁的名字 又要锁的标识 还要计数器,这里不能单纯的用key value了,要用hashmap
我们先来看整体的锁重入的流程
整个的流程都十分严谨,而且比较合理
我们来简单的捋一下
- 首先,判断锁是否存在
- 锁存在,此时判断锁上的标识是否是自己,如果是自己,那么计数器 + 1,代表又多一个人拥有此锁
- 锁不存在,那么此时就是新锁,我们应该创建一个新锁,并且置计数器为1
- 不管锁是否存在,都要设置锁的有效期,和我们自己实现不同的是,如果我们不写有效期,他会有一个默认的看门狗有效期,为30s,这里可以看我下面的源码分析 - 执行完业务了,就该到释放锁的流程了
- 先判断这个锁是不是自己,如果不是自己,那么锁就已经释放了
- 如果是自己,就让锁计数 - 1,此时再去判断此时的计数器是否到了0,如果是的化,释放锁
不管是获取锁 还是 释放锁,他们底层都是用lua脚本来实现的,使用lua脚本来实现这些命令,也是为了操作redis时命令的原子性,避免线程安全问题
lua脚本
所以我们先来看获取锁 + 释放锁的lua脚本,这样我们再看源码的时候就会有所准备
获取锁
和我们的流程一致
先是判断锁是否存在,不存在获取锁,并且设置有效期 返回1
存在的化,就去判断锁标识是不是自己,如果不是自己,获取锁失败 返回0
如果是自己,计数器 + 1,设置有效期,返回1
这个lua脚本还是很好理解的,如果你不懂lua脚本,就把这个看成是js代码,差不多的
释放锁
先是判断锁是否是自己,如果是不是自己,说明锁的主人换了,返回nil,这里也是为了解决锁误删问题
是自己锁的化,就去计数器-1,然后判断计数器是否为0,如果是0,释放锁,不是的化,就正常往下执行
源码查看
我这里的源码解读是十分粗浅的,我就是看的是一个大概,并没有非常仔细,我的想法是想大概搞懂这个原理 + 流程就ok了
先点开tryLock的代码,点到如下的是是实现类RedissonLock就是我们普遍使用的lock
点这个方法里边
这里的lua脚本是简略版的,直接用参数,但是中心思想还是一样的,所以这就是锁重入的原理
总结
我们总结一下,锁重入的原理
我们应该搞清楚一个关键的问题,如何知道是同一个人再拿锁,答案就是设置锁标识,锁上面写了你的名字相当于
所以,当我们获取锁的时候,就应该判断锁锁是不是自己的,如果是自己的,已经有锁了,我们就再锁的计数器 + 1
然后我们再释放锁的时候,也要去判断锁是不是自己,如果是自己的化,计数器 - 1
然后判断此时要不要去释放锁,如果计数器 为0了,那么就是释放速,反之就不用
锁超时 + 锁重试的原理
因为锁超时和锁重试有着千丝万缕的关系,所以一起来看是最好的
锁重试
@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);
//如果返回的是null的话,就拿到了锁返回true
// lock acquired
if (ttl == null) {
return true;
}
//此时是没有获得锁,接下来要进行重试的阶段
time -= System.currentTimeMillis() - current;
//如果此时重试的时间已经到了,就返回false
if (time <= 0) {
acquireFailed(waitTime, unit, threadId);
return false;
}
//再获取一遍
current = System.currentTimeMillis();
//订阅,订阅他人释放锁的信号
RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
//等待time的时间,如果这个时间内,都没有人过来发信号的话,就会直接失败
if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
if (!subscribeFuture.cancel(false)) {
subscribeFuture.onComplete((res, e) -> {
if (e == null) {
//取消这个订阅
unsubscribe(subscribeFuture, threadId);
}
});
}
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) {
subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(waitTime, unit, threadId);
return false;
}
}
} finally {
unsubscribe(subscribeFuture, threadId);
}
// return get(tryLockAsync(waitTime, leaseTime, unit));
}
需要注意是,再获取锁的一开始就开始计时了
比较特殊的就是这里的订阅机制,redisson如果锁失败不会一直忙等的,而是等待一段时间,这里的时间是不一定的,因为是订阅释放锁的信号,也就是当别人释放锁的时候,会发出这样的信号,这样这里就会进行重试
然后就是while true循环,去重试了,再循环的过程中,也会去维持此时的等待的剩余时间,也是基于订阅机制的
总结
总结来说, 锁重试的原理的关键在于订阅机制,订阅释放锁信号,这样很大程度上减少cpu的消耗,
然后就是比较正常的重试了,这我们都差不多能懂
锁超时
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
if (leaseTime != -1) {
return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// lock acquired 获取锁成功
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
先是这里的 ttlRemainingFuture.onComplete方法,类似于回调函数,当我们尝试去获取锁的时候,会有回调,e是异常,
如果ttlRemaining == null的化,说明获取成功了,获取锁成功之后就去更新有效期,或者说更新租约,下面就是更新的代码
private void scheduleExpirationRenewal(long threadId) {
ExpirationEntry entry = new ExpirationEntry();
//第一个参数当前锁的名称
ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
//如果有重入锁的话,那么这里的oldEntry就不是空值
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
//如果是第一次来的话,oldEntry他返回的是null
entry.addThreadId(threadId);
//更新有效期,因为是第一次来嘛
renewExpiration();
}
}
这里的EXPIRATION_RENEWAL_MAP,类似于房东一样,管理着所有的key的过期时间
如果我们是重入锁的化,这里的oldEntry就不会是空值,我们将线程id设置进去
如果是第一次来的化,就会去更新租约renewExpiration
private void renewExpiration() {
ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ee == null) {
return;
}
//定时任务,这里是个延时任务,这里的延迟时间就是internalLockLeaseTime / 3
//如果我们不写释放时间的话,internalLocakLeaseTime就是30s,那么这里的释放时间就是10s
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ent == null) {
return;
}
Long threadId = ent.getFirstThreadId();
if (threadId == null) {
return;
}
RFuture<Boolean> future = renewExpirationAsync(threadId);
future.onComplete((res, e) -> {
if (e != null) {
log.error("Can't update lock " + getName() + " expiration", e);
return;
}
if (res) {
// reschedule itself
renewExpiration();
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
这个方法就比较重要了,是看门狗的逻辑
Timeout 是定时任务,定时的时间是(看门狗的时间) / 3
也就是这个代码 internalLockLeaseTime / 3, 那么默认来说看门狗的时间是30s,那么这里默认就是10s一次,开始续约
我们还需要提前注意,只要当我们的过期时间是空的时候,才会有看门狗的存在,如果我们去看设置了过期时间的化,是不会到这里来的,我们看这里代码就懂了
回到这里的看门狗的逻辑,再timeout中的run方法就是核心代码
我们要先着眼于这个方法
这个方法就是最最最核心的方法了,已经不能再底层了,逻辑就是判断这个锁是不是自己的,然后更新有效期
再之后,回到看门狗代码
就会有一个递归的代码
意思是,不断的去进行重置
总结
按我们来看,只有当我们没有设置过期时间的时候,就会有看门狗的机制,看门狗的机制还是为了解决线程安全问题,实现了一个自动化,并且比较巧妙的是,就算我们拿到锁突然宕机了,那这里的重新去重置有效期,也不会去进行,也就是说,他会自动停下来,不用我们手动去搞
总结下来的化,看门狗这个名字挺贴切的,他会帮我们把手好过期时间的大门,然后如果我们的房子着火了的化,他还会自动的将门打开,自动化~
对于释放锁而言就会发送示范锁的消息然后去取消看门狗
主从一致性问题
正常来说,java给主机发一个锁的操作,也就是想要新开一个锁
同步还没完成,主机就宕机了,然后由于从机由哨兵模式,就会重新整一个为主机
这个问题一定是会存在的,如果说我们这里的锁再主机上设置了,还没同步到从机上,突然宕机,锁就会失效
解决办法
解决办法也很简答,我们要把获取锁成功的条件改为,所有节点都获得到这个锁才算成功
这样完全就可以根治这个问题,除非所有节点全都宕机了,那就不用先关心这个锁的问题,先把redis搞好先
但是我想了一下,不应该所有都有,为了性能 的化,我认为还是一个主节点 + 从节点都收到了这个锁的化,就算成功!
