黑马点评(五) -- 分布式锁-redission_redission使用分布式锁

1 . 分布式锁-redission功能介绍

基于setnx实现的分布式锁存在下面的问题：

重入问题：重入问题是指 获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如HashTable这样的代码中，他的方法都是使用synchronized修饰的，假如他在一个方法内，调用另一个方法，那么此时如果是不可重入的，不就死锁了吗？所以可重入锁他的主要意义是防止死锁，我们的synchronized和Lock锁都是可重入的。

不可重试：是指目前的分布式只能尝试一次，我们认为合理的情况是：当线程在获得锁失败后，他应该能再次尝试获得锁。

超时释放：我们在加锁时增加了过期时间，这样的我们可以防止死锁，但是如果卡顿的时间超长，虽然我们采用了lua表达式防止删锁的时候，误删别人的锁，但是毕竟没有锁住，有安全隐患

主从一致性： 如果Redis提供了主从集群，当我们向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题。

那么什么是Redission呢

        Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

Redission提供了分布式锁的多种多样的功能

2 . Redisson快速入门

1 . 引入依赖

<dependency>
	<groupId>org.redisson</groupId>
	<artifactId>redisson</artifactId>
	<version>3.13.6</version>
</dependency>

2 . 配置Redisson客户端

@Configuration
public class RedissonConfig {
 
    @Bean
    public RedissonClient redissonClient(){
        // 配置
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.150.101:6379")
            .setPassword("123321");
        // 创建RedissonClient对象
        return Redisson.create(config);
    }
}

3 . 使用Redisson的分布式锁

@Resource
private RedissionClient redissonClient;
 
@Test
void testRedisson() throws Exception{
    //获取锁(可重入)，指定锁的名称
    RLock lock = redissonClient.getLock("anyLock");
    //尝试获取锁，参数分别是：获取锁的最大等待时间(期间会重试)，锁自动释放时间，时间单位
    boolean isLock = lock.tryLock(1,10,TimeUnit.SECONDS);
    //判断获取锁成功
    if(isLock){
        try{
            System.out.println("执行业务");          
        }finally{
            //释放锁
            lock.unlock();
        }
        
    }   
}

tryLock方法介绍 : 

• tryLock()：它会使用默认的超时时间和等待机制。具体的超时时间是由 Redisson 配置文件或者自定义配置决定的。
• tryLock(long time, TimeUnit unit)：它会在指定的时间内尝试获取锁（等待time后重试），如果获取成功则返回 true，表示获取到了锁；如果在指定时间内（Redisson内部默认指定的）未能获取到锁，则返回 false。
• tryLock(long waitTime, long leaseTime, TimeUnit unit)：指定等待时间为watiTime，如果超过 leaseTime 后还没有获取锁就直接返回失败;

用postman进行测试:

3 .  分布式锁-redission可重入锁原理

在Lock锁中，他是借助于底层的一个voaltile的一个state变量来记录重入的状态的，比如当前没有人持有这把锁，那么state=0，假如有人持有这把锁，那么state=1，如果持有这把锁的人再次持有这把锁，那么state就会+1 ，如果是对于synchronized而言，他在c语言代码中会有一个count，原理和state类似，也是重入一次就加一，释放一次就-1 ，直到减少成0 时，表示当前这把锁没有被人持有。

在redission中，我们的也支持支持可重入锁 : 

redisson可重入锁原理 : 

按照个人理解 : 

• 对于每次进入锁，val+1,每次释放锁val-1,如果val==0,那么标识当前处于方法最外层 ;

  详见 : 19.分布式锁-Redisson的可重入锁原理_哔哩哔哩_bilibili

源码脚本 : 

redisson可重入机制测试 : 

package com.hmdp;
 
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.boot.test.context.SpringBootTest;
 
import javax.annotation.Resource;
 
@SpringBootTest
@Slf4j
public class RedissonTest {
 
        @Resource
        private RedissonClient redissonClient;
 
        private RLock lock;
        /**
         * 方法1获取一次锁
         */
        @Test
        void method1() {
            boolean isLock = false;
            // 创建锁对象
            lock = redissonClient.getLock("lock");
            try {
                isLock = lock.tryLock();
                if (!isLock) {
                    log.error("获取锁失败，1");
                    return;
                }
                log.info("获取锁成功，1");
                method2();
            } finally {
                if (isLock) {
                    log.info("释放锁，1");
                    lock.unlock();
                }
            }
        }
 
        /**
         * 方法二再获取一次锁
         */
        void method2() {
            boolean isLock = false;
            try {
                isLock = lock.tryLock();
                if (!isLock) {
                    log.error("获取锁失败, 2");
                    return;
                }
                log.info("获取锁成功，2");
            } finally {
                if (isLock) {
                    log.info("释放锁，2");
                    lock.unlock();
                }
            }
        }
 
}

核心 : 

用一个哈希结构来存线程和获取锁的次数 ;

4 . 分布式锁-redission锁重试和WatchDog机制

不可重试问题 : 获取锁一次就返回false , 没有重试机制 ;

源码跟踪理解 : 20.分布式锁-Redisson的锁重试和WatchDog机制_哔哩哔哩_bilibili

首先查看tryLock函数的参数列表 : 

• 第一个参数waitTime是获取锁的最大等待时长，如果加上这个参数 ， 不会如果获取失败就直接返回false,会在等待时间内进行获取锁的多次尝试 ，直到等待时间结束还没有获取锁成功 ，才会返回false,加了这个参数就变成了一个可重试的锁了；
• 第二个参数leaseTime是锁自动失效，释放的时间；
• 最后一个是时间的单位；

这里设置等待时间为1秒钟 ，超过一秒未成功直接返回false;

然后来追踪源码 (ctrl + alt + 左键): 

进来之后 (leaseTime没传，自动赋成-1): 

then : 

1 . 将时间转换成毫秒；

2 . 获取当前时间

3 . 获取线程id;

4 . 尝试获取锁;

then : 

then : 

1 . 首先判断释放时间是否是-1,如果传了的话，那么就不是-1,那么就会根据你传的leaseTime去执行;

2 . 反之会用getLockWatchdogTimeout()(看门狗)方法来设置你的超时时间

点进去可以看到getLockWatchDogTimeOut的超时时间是30s:

然后再进来的话，就可以发现一段 包含Lua脚本的代码 : 

如果获取锁成功，返回nil,获取锁失败，返回剩余有效期 ;

然后哦回到这里 : 

返回到最初的地方 : 

1 . 如果ttl==null表示获取锁成功 ， 直接返回true ;

2 . 否则 ， 用当前时间减去之前的时间得到获取锁消耗的时间 ;然后用time(=waitTime)来减去这个消耗的时间，如果waitTime还小于消耗的时间，那么直接返回false即可 ;

3 . 如果等待时间还有剩余，那么就继续获取当前时间，但是现在还不能够直接获取 ， 直接获取的话，可能另一线程还在处理 ， 获取失败之后立即重新获取大概率还是失败，那么就只会增加cpu的负担，那么这里用subscribe()函数来订阅脚本里面的通知 :

4 . 这是一个Future方法 ， 如果在waitTime时间内还是收不到通知，那么先释放这个订阅，然后再返回false;

5 . 如果能够得到消息，就重复执行上面获取锁的代码 : 

然后再跟踪到尝试获取锁的方法 : 

获取锁成功之后 ， 如果剩余有效期==null : 那么就要解决有效期的问题 : 

then : 

这个如果是重入锁的话，那么putIfAbsent就会保证同一个锁拿到的永远都是同一个值 ;

不管新的还是旧的，都会加入到map中，新的还会多执行一个renewExpiration操作 ;

renewExpiration :

这是一个延时任务 : 等待leaseTime的1/3后再来执行这个任务 ;

然后执行一个带脚本的方法  , 来更新有效期: 

最后会递归调用自己 ， 那么锁的有效期就会无线重置 : 

只有当锁释放的时候，才会取消跟新任务 :

那么怎么取消的呢?

直接从map中获取任务 ， 如果timeout!=null,那么取消任务 ;

总结 : 

获取锁 :

• 首先尝试获取锁 , 获取有效期ttl;

• 判断ttl是否为null：

  • 如果为null,那么代表获取锁成功，再判断leaseTime是否是-1,如果不是-1,那么也就是自己设置了一个leaseTime,也就不会去设置看门狗过期时间 ，直接返回true并结束 ;如果为-1,才会去开启watchDog,不停更新有效期，然后返回true ;

  • 如果不为null,那么表示获取锁失败，先判断当前等待时间是否大于0,小于0的话，直接返回false , 大于0的话，那么订阅并且等待释放锁的信号，再次判断等待时间是否超时，是的话就返回false,不是的话，那么再次回到尝试获取锁的步骤;

释放锁 :

• 先尝试释放锁 ， 入宫成功 ， 先发送释放锁得的消息 ， 前面的订阅任务会收到该消息 ， 然后取消watchDog,最后结束

• 如果失败，那么记录异常，并且直接结束 ;

5 .  分布式锁-redission锁的MutiLock原理

为了提高redis的可用性，我们会搭建集群或者主从，现在以主从为例 ： 

主节点处理发来的写操作， 从结点处理一些读的操作 ;

此时我们去写命令，写在主机上， 主机会将数据同步给从机，但是假设在主机还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并且选举一个slave变成master，而此时新的master中实际上并没有锁信息，此时锁信息就已经丢掉了。

        为了解决这个问题，redission提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的， 这把锁加锁的逻辑需要写入到每一个主丛节点上，只有所有的服务器都写入成功，此时才是加锁成功，假设现在某个节点挂了，那么他去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性。

测试 : 

先将另外两个结点配置进来 : 

测试类 : 

代码逻辑都一样，其它的不用改 ;

那么MutiLock 加锁原理是什么呢？笔者画了一幅图来说明

        当我们去设置了多个锁时，redission会将多个锁添加到一个集合中，然后用while循环去不停去尝试拿锁，但是会有一个总共的加锁时间，这个时间是用需要加锁的个数 * 1500ms ，假设有3个锁，那么时间就是4500ms，假设在这4500ms内，所有的锁都加锁成功， 那么此时才算是加锁成功，如果在4500ms有线程加锁失败，则会再次去进行重试.

总结 :