【Redis】4.万字文章带你深入Redisson与源码解读(建议收藏)

今天药忘吃喽~ 2024-04-01 12:41 78阅读 0赞

#### 文章目录 ####

*  1. 前言回顾
 *  2. Redisson概述
 *  3.Redisson功能介绍
 *  4. Redission的使用
 *  5. Redission可重入锁原理
 *  6. Redission可重入锁源码分析
 *  7. Redission锁重试和WatchDog机制
 *  8. Redission锁的MutiLock原理

## 1. 前言回顾 ##

在前面[【Redis】3.详解分布式锁][Redis_3.] 解决了误删锁和原子性的问题

但是不难发现，还存在一个问题，会导致锁不住。当锁的过期时间到了时候，锁就会被释放掉。这时候如果线程1因为阻塞还未向数据库插入订单数据，锁就被释放了，线程2就会获取锁，进行业务逻辑，这就导致锁不住。

因此，当过期时间到了之后，业务逻辑还没执行完的时候，我们应该给这把锁续费，就不会出现这种问题了，这个续费问题该怎么解决呢？这就需要用到Redisson了

--------------------

## 2. Redisson概述 ##

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。

它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。

Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

--------------------

## 3.Redisson功能介绍 ##

在基于setnx实现的分布式锁中，虽然解决了误删锁和原子性问题，但是还存在以下问题：

1.  **重入问题**：重入指获取锁的线程再次进入相同锁的代码块中，可以自动获取锁，从而防止了死锁的出现。常见的synchronized和Lock锁都是可重入的。
2.  **不可重试**：获取锁只尝试一次就返回false，没有重试机制。而我们希望的是，获取锁失败后可以尝试再次获取锁。
3.  **超时释放**：在加锁的时候添加了过期时间，这样有可能锁是因为超时释放，虽然使用Lua表达式解决了误删，但是代码块因为锁超时释放而没有锁住代码块，难免出现安全隐患。
4.  **主从一致性**：Redis在集群情况下，向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题。

而Redisson提供的分布式锁和多种多样的问题就可以解决上诉问题

![在这里插入图片描述][3ac20f91c0264de092e397ad5fa155ad.png_pic_center]

--------------------

## 4. Redission的使用 ##

引入依赖，这里不用redisson-spring-boot-starter是因为防止与redis的配置冲突

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

配置Redission客户端

@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);
        }
    }

Redission的分布式锁的简单使用

@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();
            }
        }   
    }

Redisson结合实际业务

@Resource
    private RedissonClient redissonClient;
    
    @Override
    public Result seckillVoucher(Long voucherId) {
        
            // 1.查询优惠券
            SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
            // 2.判断秒杀是否开始
            if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        
                // 尚未开始
                return Result.fail("秒杀尚未开始！");
            }
            // 3.判断秒杀是否已经结束
            if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
        
                // 尚未开始
                return Result.fail("秒杀已经结束！");
            }
            // 4.判断库存是否充足
            if (voucher.getStock() < 1) {
        
                // 库存不足
                return Result.fail("库存不足！");
            }
            Long userId = UserHolder.getUser().getId();
    
            RLock lock = redissonClient.getLock("lock:order:" + userId);
            //获取锁对象
            boolean isLock = lock.tryLock();
           
    		//加锁失败
            if (!isLock) {
        
                return Result.fail("不允许重复下单");
            }
            try {
        
                //获取代理对象(事务)
                IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
                return proxy.createVoucherOrder(voucherId);
            } finally {
        
                //释放锁
                lock.unlock();
            }
     }

--------------------

## 5. Redission可重入锁原理 ##

可重入锁的实际应用例子如下

线程1在method1方法获取到锁A，在执行业务代码的时候调用method2方法，在method2方法中也要尝试获取锁A，也就是持有这把锁的人要再次获得这把锁，这就是可重入锁。

@Test
    void method1() throws InterruptedException {
        
        // 尝试获取锁
        boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);
        if (!isLock) {
        
            log.error("获取锁失败 .... 1");
            return;
        }
        try {
        
            log.info("获取锁成功 .... 1");
            method2();
            log.info("开始执行业务 ... 1");
        } finally {
        
            log.warn("准备释放锁 .... 1");
            lock.unlock();
        }
    }
    void method2() {
        
        // 尝试获取锁
        boolean isLock = lock.tryLock();
        if (!isLock) {
        
            log.error("获取锁失败 .... 2");
            return;
        }
        try {
        
            log.info("获取锁成功 .... 2");
            log.info("开始执行业务 ... 2");
        } finally {
        
            log.warn("准备释放锁 .... 2");
            lock.unlock();
        }
    }

Redisson的可重入锁设计与jdk中的Lock锁设计思路差不多。

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

详细可阅读[【JUC】2.多线程锁][JUC_2.]

**在Redisson中，不再用简单的key-value来实现分布式锁。而是使用key-hashMap来实现分布式锁，hashMap中也是key-value组合，key为表示哪个线程持有这把锁，value为锁的重入次数。**

![在这里插入图片描述][a8de5e6cd4c541f7a2f6828fb7b2971e.png_pic_center]

因此，获取锁的流程也会有所改变。

线程A进来获取锁，首先判断锁是否存在，如果不存在，那么获取锁，并添加自己的线程标识，并设置锁的有效期，随后执行业务。

在执行业务的时候再次获取该锁，首先也是判断锁是否存在，很明显，锁已经存在了，**那么判断锁的标识是否是当前线程的（也就是判断这把锁是不是自己持有了）**，如果否，证明锁是被其他线程占有，获取锁失败。如果是，需要可重入，则重入次数+1，并重置锁的有效期，执行相应业务。

这里的业务执行完毕之后，首先判断锁是否是自己的（防止线程阻塞导致锁过期，因而释放掉其他线程的锁，也就是防止锁误删情况），如果是，那么锁的重入次数-1，接着判断锁的重入次数是否为0，**如果为0，证明是已经到了最外层，就可以把锁释放掉了**。

> tip：这是重置锁的有效期，是因为留下充足时间给剩下的业务执行，防止业务时间过长导致锁提前释放，造成安全问题。

同时，这一整个流程需要保证原子性，因此需要用Lua脚本。

获取锁的Lua脚本

local key = KEYS[1]; -- 锁的key
    local threadId = ARGV[1]; -- 线程唯一标识
    local releaseTime = ARGV[2]; -- 锁的自动释放时间
    -- 判断是否存在
    if(redis.call('exists', key) == 0) then
        -- 不存在, 获取锁
        redis.call('hset', key, threadId, '1'); 
        -- 设置有效期
        redis.call('expire', key, releaseTime); 
        return 1; -- 返回结果
    end;
    -- 锁已经存在，判断threadId是否是自己
    if(redis.call('hexists', key, threadId) == 1) then
        -- 不存在, 获取锁，重入次数+1
        redis.call('hincrby', key, threadId, '1'); 
        -- 设置有效期
        redis.call('expire', key, releaseTime); 
        return 1; -- 返回结果
    end;
    return 0; -- 代码走到这里,说明获取锁的不是自己，获取锁失败

释放锁的Lua脚本

local key = KEYS[1]; -- 锁的key
    local threadId = ARGV[1]; -- 线程唯一标识
    local releaseTime = ARGV[2]; -- 锁的自动释放时间-- 判断当前锁是否还是被自己持有
    if (redis.call('HEXISTS', key, threadId) == 0) then
        return nil; -- 如果已经不是自己，则直接返回
    end;
    -- 是自己的锁，则重入次数-1
    local count = redis.call('HINCRBY', key, threadId, -1);
    -- 判断是否重入次数是否已经为0 
    if (count > 0) then
        -- 大于0说明不能释放锁，重置有效期然后返回
        redis.call('EXPIRE', key, releaseTime);
        return nil;
    else  -- 等于0说明可以释放锁，直接删除
        redis.call('DEL', key);
        return nil;
    end;

--------------------

## 6. Redission可重入锁源码分析 ##

**跟踪进去tryLock( )方法**

boolean isLock = redisLock.tryLock();

**底层是一个接口，查看RessionLock的实现**

boolean tryLock();

![在这里插入图片描述][cf54b7bfcce1437ea05ab888c9250d9b.png_pic_center]

**这里调用了一个tryLockAsync()的方法**

public boolean tryLock() {
        
        return (Boolean)this.get(this.tryLockAsync());
    }

**tryLockAsync方法中调用了tryLockAsync(long threadId)，参数为当前线程ID**

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

**而tryLockAsync(long threadId)则是调用了tryAcquireOnceAsync方法**

public RFuture<Boolean> tryLockAsync(long threadId) {
        
        return this.tryAcquireOnceAsync(-1L, -1L, (TimeUnit)null, threadId);
    }

**tryAcquireOnceAsync有四个参数：**

*  waitTime：获取锁的最大等待时间（没有传默认为-1）
 *  leaseTime：锁自动释放的时间（没有传的话默认-1）
 *  unit：时间的单位（等待时间和锁自动释放的时间单位）
 *  threadId：当前线程ID

private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        
        //判断是否有锁自动释放时间
        if (leaseTime != -1L) {
        
            return this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        } else {
        
            RFuture<Boolean> ttlRemainingFuture = this.tryLockInnerAsync(
                //获取锁等待时间
                waitTime, 
                //锁自动释放时间（系统指定）
                this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), 
                //时间单位
                TimeUnit.MILLISECONDS, 
                //线程ID
                threadId, 
                .EVAL_NULL_BOOLEAN);
            ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        
                if (e == null) {
        
                    if (ttlRemaining) {
        
                        this.scheduleExpirationRenewal(threadId);
                    }
    
                }
            });
            return ttlRemainingFuture;
        }
    }

在tryAcquireOnceAsync方法中，首先要先判断有没有锁自动释放时间（leaseTime），如果有，则会使用有的那个纸，否则会给一个默认值。

接着会调用tryLockInnerAsync方法。

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        
        //将锁自动释放时间记录在一个本地的成员变量中
        this.internalLockLeaseTime = unit.toMillis(leaseTime);
        return this.evalWriteAsync(
            this.getName(), 
            LongCodec.INSTANCE, 
            command, 
            //Lua脚本，获取锁失败的话返回pttl（和ttl差不多，只是单位不一样，ttl是秒，pttl是毫秒），也就是这个key的剩余有效期
            "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(this.getName()), 
            this.internalLockLeaseTime, 
            this.getLockName(threadId)
        );
    }

在tryLockInnerAsync方法，可以发现这里面有一段Lua脚本，其作用和上面第5点的Lua脚本一样。

这就是获取锁的源码解读。

接下来看看释放锁的源码。

同样的，调用的是Lock类下的unlock方法

void unlock();

查看RedissionLock实现类下的unlock方法

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

unlock方法，主要是看unlockAsync方法，参数为当前线程ID

public RFuture<Void> unlockAsync(long threadId) {
        
        RPromise<Void> result = new RedissonPromise();
        RFuture<Boolean> future = this.unlockInnerAsync(threadId);
        future.onComplete((opStatus, e) -> {
        
            this.cancelExpirationRenewal(threadId);
            if (e != null) {
        
                result.tryFailure(e);
            } else if (opStatus == null) {
        
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + this.id + " thread-id: " + threadId);
                result.tryFailure(cause);
            } else {
        
                result.trySuccess((Object)null);
            }
        });
        return result;
    }

这里继续跟进unlockInnerAsync方法，参数还是线程ID

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

unlockInnerAsync方法下同样有一段Lua脚本，这段Lua脚本和上面第5点释放锁的Lua脚本一样。

--------------------

## 7. Redission锁重试和WatchDog机制 ##

尝试获取锁的tryLock方法有以下几种重载方法

boolean tryLock();

boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException

三个参数：

1.  waitTime：获取锁的最大等待时间（没有传默认为-1）
2.  leaseTime：锁自动释放的时间（没有传的话默认-1）
3.  unit：时间的单位（等待时间和锁自动释放的时间单位）

**假如没传leaseTime，则time就是指获取锁的最大等待时间（没有传默认为-1），而且leaseTime会给默认值**

跟进第二种tryLock方法

public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        
        return this.tryLock(waitTime, -1L, unit);
    }

这里是调用了一个tryLock方法，由于没有传leaseTime，所以默认设置为-1

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        
        //将等待时间转化为毫秒
        long time = unit.toMillis(waitTime);
        //获取当前时间
        long current = System.currentTimeMillis();
        //获取线程ID
        long threadId = Thread.currentThread().getId();
        //尝试获取锁
        //这里会有两种情况，一种是nil代表获取锁成功，一种是该key的剩余有效期代表回去锁失败
        Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
        if (ttl == null) {
        
            //代表获取锁成功
            return true;
        } else {
        
            //获取锁失败，要尝试再次获取
            //使用当前时间减去之前记录当前时间，也就是获取锁消耗掉的时间
            //再用最长等待时间，减去获取锁消耗的时间，得到的结果就是剩余等待时间
            time -= System.currentTimeMillis() - current;
            if (time <= 0L) {
        
                //剩余等待时间小于0，获取锁失败
                this.acquireFailed(waitTime, unit, threadId);
                return false;
            } else {
        
                //剩余等待时间大于0，就可以继续尝试
                //获取当前的时间，准备进行尝试
                //但是不是立即尝试，因此刚才获取失败，如果立刻尝试，那个获取锁的线程大概率是没有完成业务释放锁的
                current = System.currentTimeMillis();
                //于是可以订阅别人是释放锁的信号——this.subscribe(threadId);
                //可以这样，是因为在释放锁的时候会发布一条释放锁的通知
                RFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);
                //等待 这剩余最长等待时间 还没有释放锁的话，那么获取锁失败
                if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
        
                    //剩余最长等待时间内 其他线程还未释放锁 获取锁失败
                    if (!subscribeFuture.cancel(false)) {
        
                        subscribeFuture.onComplete((res, e) -> {
        
                            if (e == null) {
        
                                //会先取消订阅
                                this.unsubscribe(subscribeFuture, threadId);
                            }
    
                        });
                    }
                    this.acquireFailed(waitTime, unit, threadId);
                    //获取锁失败
                    return false;
                } else {
        
                    boolean var16;
                    try {
        
                        //获取当前时间，当前时间减去之前记录的当前时间，得到的是获取锁成功花费的时间
                        //之前剩余的最长等待时间-获取锁成功花费的时间
                        time -= System.currentTimeMillis() - current;
                        if (time <= 0L) {
        
                            //剩余最长等待时间小于0，获取锁失败
                            this.acquireFailed(waitTime, unit, threadId);
                            boolean var20 = false;
                            return var20;
                        }
    					//这一次等待结束，最长等待时间依然有剩余
                        do {
        
                            //当前时间
                            long currentTime = System.currentTimeMillis();
                            //进行重试，尝试获取锁
                            //获取锁失败的返回值是该key的剩余过期时间
                            ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
                            if (ttl == null) {
        
                                //获取锁成功
                                var16 = true;
                                return var16;
                            }
    						//计算获取锁花费的时间
                            //用剩余等待时间减去获取锁的时间
                            time -= System.currentTimeMillis() - currentTime;
                            if (time <= 0L) {
        
                                //剩余最长等待时间小于0，获取锁失败
                                this.acquireFailed(waitTime, unit, threadId);
                                var16 = false;
                                return var16;
                            }
                            
                            
                            /**
                            
                            （重试机制）
                            下面这段代码设计得非常巧妙
                            它并不是无休止得循环等
                            而是先等待另外一个线程释放锁，再进程重试
                            这样设计减少了CPU的浪费
                            
                            **/
                            
                            
    						//还剩下最长等待时间的大于0
                            currentTime = System.currentTimeMillis();
                           	//ttl < time：也就是该key的剩余过期时间小于剩余的最长等待时间
                            if (ttl >= 0L && ttl < time) {
        
                               	//使用信号量，因为释放锁的人会释放一个信号，这边就尝试获取信号
                                //如果是指定等待时间内，拿到了这个信号量返回true
                                //否则返回false
                                //那么只需要等待该key剩余等待时间即可
                                //消息ding
                                ((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                            } else {
        
                                //如果ttl >= time
                                //那么只需要等待time，也就是剩余的最长等待时间
                                //如果这段时间都还没等到信号量，那么就证明失败了
                                //订阅
                                ((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                            }
                            
                            
                            
                           
    						//重置剩余的最长的等待时间
                            time -= System.currentTimeMillis() - currentTime;
                        } while(time > 0L);
    					//既然能从循环出来，证明time肯定小于等于0
                        //也就是最长等待时间都用完了，获取锁失败
                        this.acquireFailed(waitTime, unit, threadId);
                        var16 = false;
                    } finally {
        
                        //取消订阅
                        this.unsubscribe(subscribeFuture, threadId);
                    }
    
                    return var16;
                }
            }
        }
    }

跟进tryAcquire发现，其是调用一个tryAcquireAsync方法

private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        
        return (Long)this.get(this.tryAcquireAsync(waitTime, leaseTime, unit, threadId));
    }

tryAcquireAsync方法就是上面解读Redission的可重入锁的源码调用到的一个方法

**上面有说到，没传leaseTime（自动释放锁时间）的话，就会给一个默认值，这个默认值就是getLockWatchdogTimeout()，也就是看门狗超时时间**

**这个看门狗超时时间是30\*1000毫秒，也就是30秒**

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        
        if (leaseTime != -1L) {
        
            return this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        } else {
        
            //如果获取锁失败，返回的结果是这个key的剩余有效期
            RFuture<Long> ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
            //上面获取锁回调成功之后，执行这代码块的内容
            ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        
                if (e == null) {
        
                    //剩余有效期为null
                    if (ttlRemaining == null) {
        
                        //这个函数是解决最长等待有效期的问题
                        this.scheduleExpirationRenewal(threadId);
                    }
    
                }
            });
            return ttlRemainingFuture;
        }
    }

如果获取锁返回的锁key的剩余有效期的时间为null的时候（也就是获取锁成功），就会解决最长等待剩余有效期的问题

private void scheduleExpirationRenewal(long threadId) {
        
        ExpirationEntry entry = new ExpirationEntry();
        //这里EntryName是指锁的名称
        ExpirationEntry oldEntry = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
        if (oldEntry != null) {
        
            //重入
            //将线程ID加入
            oldEntry.addThreadId(threadId);
        } else {
        
            //将线程ID加入
            entry.addThreadId(threadId);
            //续约
            this.renewExpiration();
        }
    
    }

这里的entryName是指锁的名称，因为创建RedissonLock类的时候的参数name就是锁的名称

![在这里插入图片描述][07db72674e5e4b09a8be5ee8ada02c8c.png_pic_center]

因此，一个锁就对应自己的一个ExpirationEntry类

这样，如果是第一次进入的时候，这里放进去的就是一个全新的ExpirationEntry类，也就是当前锁，返回值就是null

如果是重入，那么putIfAbsent函数就不会执行，返回值就是之前旧的ExpirationEntry类，也就是第一次进来创建的ExpirationEntry类。这一步保证了无论重入几次，拿到的都是同一把锁。

如果是第一次，那么需要进行续约操作，也就是给最长等待有效期续约。

private void renewExpiration() {
        
        //先从map里得到这个ExpirationEntry
        ExpirationEntry ee = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
        if (ee != null) {
        
            //这个是一个延迟任务
            Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        
                //延迟任务内容
                public void run(Timeout timeout) throws Exception {
        
                    //拿出ExpirationEntry
                    ExpirationEntry ent = (ExpirationEntry)RedissonLock.EXPIRATION_RENEWAL_MAP.get(RedissonLock.this.getEntryName());
                    if (ent != null) {
        
                        //从ExpirationEntry拿出线程ID
                        Long threadId = ent.getFirstThreadId();
                        if (threadId != null) {
        
                            //调用renewExpirationAsync方法刷新最长等待时间
                            RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
                            future.onComplete((res, e) -> {
        
                                if (e != null) {
        
                                    RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", e);
                                } else {
        
                                    if (res) {
        
                                        //renewExpirationAsync方法执行成功之后，进行递归调用，调用自己本身函数
                                        //那么就可以实现这样的效果
                                        //首先第一次进行这个函数，设置了一个延迟任务，在10s后执行
                                        //10s后，执行延迟任务的内容，刷新有效期成功，那么就会再新建一个延迟任务，刷新最长等待有效期
                                        //这样这个最长等待时间就会一直续费
                                        RedissonLock.this.renewExpiration();
                                    }
    
                                }
                            });
                        }
                    }
                }
            }, 
                                                                                  //这是锁自动释放时间，因为没传，所以是看门狗时间=30*1000
                                                                                  //也就是10s
                                                                                  this.internalLockLeaseTime / 3L, 
                                                                                  //时间单位
                                                                                  TimeUnit.MILLISECONDS);
            //给当前ExpirationEntry设置延迟任务
            ee.setTimeout(task);
        }
    }

重置锁的有效期的关键方法是renewExpirationAsync

使用了一段Lua脚本来重置

protected RFuture<Boolean> renewExpirationAsync(long threadId) {
        
        return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('pexpire', KEYS[1], ARGV[1]); return 1; end; return 0;", Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
    }

上述就是加锁的源码解读

接下来看看释放锁的源码

释放主要看unlockAsync方法。

unlockAsync方法这里有一个unlockInnerAsync。

当这个unlockInnerAsync执行完之后，首先会取消更新任务。

public RFuture<Void> unlockAsync(long threadId) {
        
        RPromise<Void> result = new RedissonPromise();
        RFuture<Boolean> future = this.unlockInnerAsync(threadId);
        future.onComplete((opStatus, e) -> {
        
            //取消锁更新任务
            this.cancelExpirationRenewal(threadId);
            if (e != null) {
        
                result.tryFailure(e);
            } else if (opStatus == null) {
        
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + this.id + " thread-id: " + threadId);
                result.tryFailure(cause);
            } else {
        
                result.trySuccess((Object)null);
            }
        });
        return result;
    }

void cancelExpirationRenewal(Long threadId) {
        
        //获得当前这把锁的任务
        ExpirationEntry task = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
        if (task != null) {
        
            //当前锁的延迟任务不为空，且线程id不为空
            if (threadId != null) {
        
                //先把线程ID去掉
                task.removeThreadId(threadId);
            }
    
            if (threadId == null || task.hasNoThreads()) {
        
                //然后取出延迟任务
                Timeout timeout = task.getTimeout();
                if (timeout != null) {
        
                    //把延迟任务取消掉
                    timeout.cancel();
                }
    			//再把ExpirationEntry移除出map
                EXPIRATION_RENEWAL_MAP.remove(this.getEntryName());
            }
    
        }
    }

这里就把定时任务删除得干干净净

这里整个锁的释放就完成了。

流程图如下：

![在这里插入图片描述][ef37da345ef4430ea5f0308b5e06513c.png_pic_center]

--------------------

## 8. Redission锁的MutiLock原理 ##

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

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

![在这里插入图片描述][1bce8920469542b29f7b20256e536dcb.png_pic_center]

为了解决这个问题，redission提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的， 这把锁加锁的逻辑需要写入到每一个主丛节点上，只有所有的服务器都写入成功，此时才是加锁成功。假设现在某个节点挂了，有一个线程乘虚而入，想要获取锁，那么这个线程虽然再第一个节点能获取锁，但是只有再所有主节点中获取到锁，才算成功获取锁，因为其他主节点的都是被原来的线程占有，乘虚而入的线程无法获取另外两个节点的锁，因此获取锁失败。

![在这里插入图片描述][dc7e6de47a58498694e84d7447a1bf8b.png_pic_center]

**Redission锁的MutiLock的使用**

配置Redission的客户端

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

创建联锁

@Resource
    private RedissonClient redissonClient;
    
    @Resource
    private RedissonClient redissonClient2;
    
    @Resource
    private RedissonClient redissonClient3;
    
    private RLock lock;
    
    @BeforeEach
    void setUp() {
        
        
        RLock lock1 = redissonClient.getLock("order");
        RLock lock2 = redissonClient2.getLock("order");
        RLock lock3 = redissonClient3.getLock("order");
        
        lock = redissonClient.getMultiLock(lock1, lock2, lock3);
    }

进入getMultiLock方法

public RLock getMultiLock(RLock... locks) {
        
        return new RedissonMultiLock(locks);
    }

这里面是new一个RedissonMultiLock

final List<RLock> locks = new ArrayList();
    
    public RedissonMultiLock(RLock... locks) {
        
        if (locks.length == 0) {
        
            throw new IllegalArgumentException("Lock objects are not defined");
        } else {
        
            this.locks.addAll(Arrays.asList(locks));
        }
    }

**可以看到这里锁添加到一个集合中，将来获取锁的时候，会将集合里面的锁都获取一边，只有都获取成功，才算获取到锁。**

拿到锁之后，锁的使用和之前一样，加锁、释放锁。

**讲述完使用之后，下面开始分析一下源码**。

tryLock方法的两个参数：

1.  waitTime：最长等待时间
2.  unit：时间单位

public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        
        return this.tryLock(waitTime, -1L, unit);
    }

**接下来进入this.tryLock(waitTime, -1L, unit)分析**

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        
        long newLeaseTime = -1L;
        //判断传入参数有没有自动释放锁时间
        if (leaseTime != -1L) {
        
            //判断传入参数是否有最长等待时间
            if (waitTime == -1L) {
        
                //没有传的话，说明只想获取一次，不需要进行重试
                //因此自动释放锁时间多久，就用多久
                newLeaseTime = unit.toMillis(leaseTime);
            } else {
        
                //想要进行重试
                //用最长等待时间✖2
                //因为重试时间比较久，避免自动释放时间小于最长等待时间导致还没重试完，锁就释放了
                newLeaseTime = unit.toMillis(waitTime) * 2L;
            }
        }
    
        long time = System.currentTimeMillis();
        //剩余等待时间
        long remainTime = -1L;
        if (waitTime != -1L) {
        
            //最长等待时间有传进来，就会使用waitTime代替remainTime
            remainTime = unit.toMillis(waitTime);
        }
    	//获得锁等待时间
        //锁等待时间和剩余等待时间是一样的
        long lockWaitTime = this.calcLockWaitTime(remainTime);
        //失败锁的限制。默认为0
        int failedLocksLimit = this.failedLocksLimit();
        //这里保存的是获取成功的锁
        List<RLock> acquiredLocks = new ArrayList(this.locks.size());
        //这里是遍历需要获取的锁
        ListIterator<RLock> iterator = this.locks.listIterator();
    
        while(iterator.hasNext()) {
        
            RLock lock = (RLock)iterator.next();
    
            boolean lockAcquired;
            try {
        
                if (waitTime == -1L && leaseTime == -1L) {
        
                    //这里就是没参数，所以调用的是空参的tryLock
                    lockAcquired = lock.tryLock();
                } else {
        
                    //传了参数
                    long awaitTime = Math.min(lockWaitTime, remainTime);
                    //进行获取锁操作
                    lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
                }
            } catch (RedisResponseTimeoutException var21) {
        
                this.unlockInner(Arrays.asList(lock));
                lockAcquired = false;
            } catch (Exception var22) {
        
                lockAcquired = false;
            }
    
            if (lockAcquired) {
        
                //获取锁成功，将锁放进集合中
                acquiredLocks.add(lock);
            } else {
        
                //获取锁失败
                //判断锁的总数量-已经获取的锁的数量是否等于锁失败的上限
                if (this.locks.size() - acquiredLocks.size() == this.failedLocksLimit()) {
        
                    break;
                }
    			
                if (failedLocksLimit == 0) {
        
                    //将已经拿到的锁释放掉
                    this.unlockInner(acquiredLocks);
                    if (waitTime == -1L) {
        
                        //最长等待时间为0，说明不想重试
                        return false;
                    }
    				//需要重试
                    failedLocksLimit = this.failedLocksLimit();
                    //将之前获取到的锁释放掉
                    acquiredLocks.clear();
    				//把迭代器往前迭代，也就是从第一个开始
                    while(iterator.hasPrevious()) {
        
                        iterator.previous();
                    }
                } else {
        
                    --failedLocksLimit;
                }
            }
    		//判断剩余等待时间是不是-1
            if (remainTime != -1L) {
        
                //如果剩余等待时间不为-1
                //那么用剩余等待时间减去获取锁时间
                //得到现在剩余的时间
                remainTime -= System.currentTimeMillis() - time;
                time = System.currentTimeMillis();
                if (remainTime <= 0L) {
        
                    //如果现在剩余等待时间小于0，说明已经耗尽了等待时间
                   	//获取锁失败，那么将之前获取的锁释放掉
                    this.unlockInner(acquiredLocks);
                    //获取锁超时，获取锁失败
                    return false;
                }
            }
        }
    	//上面这个循环结束，证明获取锁成功
        //在返回true之前还需要进行以下操作
        if (leaseTime != -1L) {
        
            //锁的自动释放时间不是-1，说明有指定锁的自动释放时间
            List<RFuture<Boolean>> futures = new ArrayList(acquiredLocks.size());
            Iterator var24 = acquiredLocks.iterator();
    		//遍历每一把锁
            while(var24.hasNext()) {
        
                RLock rLock = (RLock)var24.next();
                //给每一个把锁设置有效期，也就是重置有效期
                /**
                
                为什么要这样做呢？
                因为当获取到第一把锁的时候，有效期就开始倒计时了
                因此第一把锁的剩余有效期一定会比最后一把锁的剩余有效期要短
                这样就会出现有些锁释放，有些还没释放的情况
                
                为什么要指定锁释放时间的时候才进行这操作？
                因为不指定的时候会触发看门狗机制，有效期会自动去续费，不需要我们操作
                
                
                **/
                RFuture<Boolean> future = ((RedissonLock)rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
                futures.add(future);
            }
    
            var24 = futures.iterator();
    
            while(var24.hasNext()) {
        
                RFuture<Boolean> rFuture = (RFuture)var24.next();
                rFuture.syncUninterruptibly();
            }
        }
    
        return true;
    }

以上就是Redission锁的MutiLock原理的源码分析

通过源码分析，不难看出，**MutiLock的缺点就是运维成本高，实现复杂**

**如有错误，欢迎指教！！！！！！**

--------------------

> 参考： [黑马程序员Redis入门到实战教程，深度透析redis底层原理+redis分布式锁+企业解决方案+黑马点评实战项目][Redis_redis_redis]

--------------------

[Redis_3.]: https://blog.csdn.net/weixin_51146329/article/details/127588196?spm=1001.2014.3001.5501
[3ac20f91c0264de092e397ad5fa155ad.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/61b475861b184f0c877bb2d3a73c2763.png
[JUC_2.]: https://blog.csdn.net/weixin_51146329/article/details/127139006?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522166710076816782417072282%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=166710076816782417072282&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~rank_v31_ecpm-2-127139006-null-null.nonecase&utm_term=%E5%8F%AF%E9%87%8D%E5%85%A5&spm=1018.2226.3001.4450
[a8de5e6cd4c541f7a2f6828fb7b2971e.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/684ce98660e8489abe5c21647d8265c6.png
[cf54b7bfcce1437ea05ab888c9250d9b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/19661bb585eb4aa6b9b3bf82ec8b4a73.png
[07db72674e5e4b09a8be5ee8ada02c8c.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/55ffd01e294b45ac8a9c00e21ec64234.png
[ef37da345ef4430ea5f0308b5e06513c.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/a807f5210b7a4cf4bbd4b080fc726fbc.png
[1bce8920469542b29f7b20256e536dcb.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/07d9e6f5c8c7450ca1db041992e95b33.png
[dc7e6de47a58498694e84d7447a1bf8b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/7ef9ed860cd94150bea08794ec1a024d.png
[Redis_redis_redis]: https://www.bilibili.com/video/BV1cr4y1671t