当前位置:网站首页>Redis分布式锁

Redis分布式锁

2022-08-11 05:21:00 ?671

引言

随着业务发展的需要,原单体单机部署的系统被演化成分布式集群系统后,由于分布式系统多线程、多进程并且分布在不同机器上,这将使原单机部署情况下的并发控制锁策略失效,单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要-种跨JVM的互斥机制来控制共享资源的访问,这就是分布式锁要解决的问题!

一、分布式锁的概念

分布式锁指的是,所有服务中的所有线程都去获得同一把锁,但只有一个线程可以成功的获得锁,其他没有获得锁的线程必须全部等待,等到获得锁的线程释放掉锁之后获得了锁才能进行操作。Redis官网中,set key value有个带有NX参数的命令,这是一个原子性加锁的命令,指的是此key没有被lock时,当前线程才能加锁,如果已经被占用,就不能加锁。

                                                                   图1

 二、分布式锁在java代码中的优化过程演示

在虚拟机开启redis服务,设置num为0,根据图1,可以初步写出如下java代码:

​
@GetMapping("testLock")  
public void testLock(){
//1获取锁,setne
Boolean lock =redisTemplate.opsForValue().setIfAbsent( k: “lock", v: "111");
//2获取锁成功、查询num的值 
if(lock){
Object value =redisTemplate.opsForValue().get("num");
//2.1判断num为空return
if(StringUtils.isEmptv(value)){
return:}
//2.2有值就转成成int
int num = Integer.parseInt(s:value+"");
/12.3把redis的num加1
redisTemplate.opsForValue().set("num”,++num);
//2.4释放锁
redisTemplate. delete( key:“lock");
}}else{
//3获取锁失败、每隔0.1秒再获取 
try{
Thread. sleep( millis: 100): 
testLock();
} catch (InterruptedException e) {
e.printStackTraceO;
}

​

 在上述代码中,看似没有问题,但是如果在上锁之后,释放锁之前的代码的执行出现问题,那么就无法释放锁了,所以此时应该在步骤1获取锁的时候给锁设置一个过期时间,如下代码:

​
@GetMapping("testLock")  
public void testLock(){
//1获取锁,setne
Boolean lock =redisTemplate.opsForValue().setIfAbsent( k: “lock", v: "111",3, TimeUnit. SECONDS,);
//2获取锁成功、查询num的值 
if(lock){
Object value =redisTemplate.opsForValue().get("num");
//2.1判断num为空return
if(StringUtils.isEmptv(value)){
return:}
//2.2有值就转成成int
int num = Integer.parseInt(s:value+"");
/12.3把redis的num加1
redisTemplate.opsForValue().set("num”,++num);
//2.4释放锁
redisTemplate. delete( key:“lock");
}}else{
//3获取锁失败、每隔0.1秒再获取 
try{
Thread. sleep( millis: 100): 
testLock();
} catch (InterruptedException e) {
e.printStackTraceO;
}

​

此时也会有一个问题,什么问题呢,接下来我举个例子

如上图所示,设置锁10秒过期,a进行操作,先上锁,然后进行具体操作,但是此时服务器突然出现卡顿,导致a的具体操作还没完成,锁就先过期自动释放。

接下来,锁一旦被释放,就被b抢占了,然后进行具体操作,但是此时a操作的服务器反应过来了,进行了锁的释放,就会导致操作b的锁被操作a释放了,进而导致锁又再次被其他操作抢占。进而导致这种混乱的场面。

因此,我们就得保证a操作只能释放自己的锁而不能去释放其他操作持有的锁,此时,我们就可以在每个操作进行上锁时,给锁设置一个uuid,然后在完成具体操作后,在java代码中通过判断该锁的uuid是不是自己操作时的uuid来判断是否释放锁了,代码如下:


@GetMapping("testLock")  
public void testLock(){
//1获取锁,setne
String uuid=UUID.randomUUID().toString();
Boolean lock =redisTemplate.opsForValue().setIfAbsent( “lock", uuid,3, TimeUnit. SECONDS,);
//2获取锁成功、查询num的值 
if(lock){
Object value =redisTemplate.opsForValue().get("num");
//2.1判断num为空return
if(StringUtils.isEmptv(value)){
return:}
//2.2有值就转成成int
int num = Integer.parseInt(s:value+"");
/12.3把redis的num加1
redisTemplate.opsForValue().set("num”,++num);
//2.4释放锁
String lockUuid=(String)redisTemplate.opsForValue().get("lock");
if(lockUuid.equals(uuid)){
redisTemplate. delete("lock");
}
}}else{
//3获取锁失败、每隔0.1秒再获取 
try{
Thread. sleep( millis: 100): 
testLock();
} catch (InterruptedException e) {
e.printStackTraceO;
}
}

​

此时,你可能会觉得这下应该没有问题了吧,那只能说你太单纯了,不了解社会的险恶,上述代码还是会出现问题,下面我就来解释一下吧

当a操作进行上锁,完成具体操作后,然后进行uuid的比较,判断得锁的uuid与自己操作时设置的uuid一致,进行释放锁的操作,但如果在正要释放锁而没有释放锁的时候,锁刚好过期,这就导致锁会被b操作抢占,然后进行具体操作,当操作a完成了对锁的释放,也就是意味着操作b的锁被操作a释放了。

为了解决这一问题,我们就得保证在uuid判断完成后进行解锁这一过程具有原子性,即在此过程中不能被其他操作抢占,这时,我们就需要用到lua脚本了。

那么lua脚本又什么用呢?它可以将复杂的或者多步的redis操作,写为一个脚本,一次提交给redis执行,减少反复连接 redis 的次数。提升性能。lua脚本是类似 redis 事务,有一定的原子性,不会被其他命令插队,可以完成一些 redis 事务性的操作。但是注意 redis的lua脚本功能,只有在Redis2.6 以上的版本才可以使用。

因此,我们只需将判断uuid和释放锁的这一过程写成lua脚本(使这一过程具有原子性)就可以解决问题了,代码如下:

​@GetMapping("testLock")  
public void testLock(){
//1获取锁,setne
String uuid=UUID.randomUUID().toString();
Boolean lock =redisTemplate.opsForValue().setIfAbsent( “lock", uuid,3, TimeUnit. SECONDS,);
//2获取锁成功、查询num的值 
if(lock){
Object value =redisTemplate.opsForValue().get("num");
//2.1判断num为空return
if(StringUtils.isEmptv(value)){
return:}
//2.2有值就转成成int
int num = Integer.parseInt(s:value+"");
/12.3把redis的num加1
redisTemplate.opsForValue().set("num”,++num);
//2.4释放锁
String lockUuid=(String)redisTemplate.opsForValue().get("lock");
/*使用lua脚本来锁*/
//定义lua脚本
String script ="if redis.call('get',KEYS[1])== ARGV[1] then return redis.call('del’ KEYS[1]) else return 0 end";
//使用redis执行lua执行
DefaultRedisSeript<Long> redisSeript=new DefaultRedisSeript<>(): redisScript.setScriptText(script);
//设置一下返回值类型为Long
//因为删除判断的时候,返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型
//那么返回字符串与会有发生错误。
redisSeript.setResultType(Long.class);
//第一个要是script脚本,第二个需要判断的key,第三个就是key所对应的值。 redisTemplate.execute(redisScript.Arravs.aslist(locKev),uuid);
}else{
//3获取锁失败、每隔0.1秒再获取 
try{
Thread. sleep( millis: 100): 
testLock();
} catch (InterruptedException e) {
e.printStackTraceO;
}
}

​

​

原网站

版权声明
本文为[?671]所创,转载请带上原文链接,感谢
https://blog.csdn.net/qq_56370103/article/details/126275206