当前位置：首页 > news >正文

Redisson分布式锁原理 + 运用记录

news 文章来源：https://blog.csdn.net/weixin_52232901/article/details/136091570 2024/9/22 21:34:31

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 {@Beanpublic 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了,那么就是释放速,反之就不用

锁超时 + 锁重试的原理

因为锁超时和锁重试有着千丝万缕的关系,所以一起来看是最好的

锁重试

    @Overridepublic 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 acquiredif (ttl == null) {return true;}//此时是没有获得锁,接下来要进行重试的阶段time -= System.currentTimeMillis() - current;//如果此时重试的时间已经到了,就返回falseif (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 acquiredif (ttl == null) {return true;}//判断是否超时time -= System.currentTimeMillis() - currentTime;if (time <= 0) {acquireFailed(waitTime, unit, threadId);return false;}// waiting for messagecurrentTime = 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他返回的是nullentry.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,那么这里的释放时间就是10sTimeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic 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 itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}