redisson分布式锁的源码解析-元一软件

一、前言分布式锁在实际工作中的应用还是比较多的其实现方式也有很多种常见的有基于数据库锁、基于zookeeper、基于redis的今天我们来讲下基于redis实现的分布式锁。redisson是一个redis客户端框架提供了分布式锁的功能特性这里我们通过解析redisson的源码来分析它是如何基于redis来实现分布式锁的二、源码解析2.1 样例代码下面是一个分布式锁的简单样例代码// 初始化配置创建Redisson客户端 Config confignew Config();config.setCodec(new JsonJacksonCodec()).useSingleServer().setAddress(redis://192.168.10.131:6379);RedissonClient clientRedisson.create(config);// 获取分布式锁 RLock lockclient.getLock(myLock);lock.lock();System.out.println(Thread.currentThread().getId(): 获取到分布式锁);try{Thread.sleep(60*1000);}catch(Exception e){e.printStackTrace();}finally{// 解锁 lock.unlock();}上面的样例代码比较简单通过redisson客户端获取一个分布式锁该分布式锁的key为myLock睡眠60秒之后释放锁。这里比较重要的是lock()方法该方法是获取锁的具体步骤所以接下来详细解析一下该方法2.2 整体流程获取锁的流程图如下具体流程为第一次尝试获取锁如果获取到锁直接返回。如果未获取到锁返回锁的剩余过期时间ttl当未获取到锁时订阅频道redisson_lock__channel:{myLock}订阅该频道的作用是当该分布式锁被其他拥有者所释放时会往该订阅频道发送一个解锁消息UNLOCK_MESSAGE这时当前等待该分布式锁的线程会中断等待并再次尝试获取锁开启死循环尝试获取锁如果未获取到锁拿到锁的剩余过期时间并等待该锁的剩余过期时间中间过程中如果订阅频道有解锁消息UNLOCK_MESSAGE会提前中断等待继续循环直到获取锁退出循环获取到锁之后取消订阅频道源码如下private void lock(long leaseTime, TimeUnit unit, boolean interruptibly)throws InterruptedException{long threadIdThread.currentThread().getId();//1、第一次尝试获取锁ttl为null表示获取到锁直接return Long ttltryAcquire(-1, leaseTime, unit, threadId);if(ttlnull){return;}//2、订阅频道redisson_lock__channel:{myLock}CompletableFutureRedissonLockEntryfuturesubscribe(threadId);pubSub.timeout(future);RedissonLockEntry entry;if(interruptibly){entrycommandExecutor.getInterrupted(future);}else{entrycommandExecutor.get(future);}try{//3、开启循环while(true){// 再次尝试获取锁ttl为null表示获取到锁退出循环 ttltryAcquire(-1, leaseTime, unit, threadId);if(ttlnull){break;}// 如果ttl大于等于0if(ttl0){try{// 等待ttl时间 或者 接收到解锁消息 entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);}catch(InterruptedException e){...}}else{// 如果ttl小于0说明该锁未设置过期时间等待接收解锁消息if(interruptibly){entry.getLatch().acquire();}else{entry.getLatch().acquireUninterruptibly();}}}}finally{// 退出订阅频道redisson_lock__channel:{myLock}unsubscribe(entry, threadId);}}2.3 锁的获取那么如何表示当前线程获取到锁redisson中的分布式锁实质上是个hash结构的数据假设锁的名称为myLock那么当某个线程获取到锁之后会在这个hash结构里设置一个hashkey其为【连接管理器id】 : 【线程id】如下图redisson通过执行lua脚本来获取锁lua脚本如下// 如果锁不存在则成功获取到锁设置锁的过期时间并返回nilif(redis.call(exists, KEYS[1])0)thenredis.call(hincrby, KEYS[1], ARGV[2],1);redis.call(pexpire, KEYS[1], ARGV[1]);returnnil;end;// 如果锁已存在判断是否是当前线程已经获取到如果是对应的值加1if(redis.call(hexists, KEYS[1], ARGV[2])1)thenredis.call(hincrby, KEYS[1], ARGV[2],1);redis.call(pexpire, KEYS[1], ARGV[1]);returnnil;end;// 否则表示未获取到锁返回锁的过期时间returnredis.call(pttl, KEYS[1]);该lua脚本的主要作用是如果锁不存在则成功获取到锁设置锁的过期时间lockWatchdogTimeout默认是30秒并返回nil如果锁已存在判断是否是当前线程已经获取到如果是对应的值加1否则表示未获取到锁返回锁的过期时间这里的第一步为什么要设置锁的过期时间其实是为了当锁的拥有者挂了之后避免锁一直存在导致其他应用永远无法获取到锁。2.4 锁续期那么既然锁设置了过期时间那很自然地想到如果在锁过期的这段时间内拥有锁的线程还未执行完业务逻辑这时锁自动过期导致其他应用也获取到了锁从而产生逻辑错误。所以引入了锁续期。当获取到锁时redisson会启动一个看门狗该看门狗每隔 lockWatchdogTimeout / 3秒续期一次锁假设lockWatchdogTimeout默认为30秒则每隔10秒续期锁源码如下private voidrenewExpiration(){... //1、创建一个10秒后执行的延迟任务 Timeout taskcommandExecutor.getConnectionManager().newTimeout(newTimerTask(){Override public void run(Timeouttimeout)throws Exception{... //2、执行续期锁的lua脚本 CompletionStageBooleanfuturerenewExpirationAsync(threadId);future.whenComplete((res, e)-{... //3、res为true代表锁续期成功重新调用该方法继续创建延迟任务 // false表示锁续期失败if(res){renewExpiration();}else{cancelExpirationRenewal(null);}});}}, internalLockLeaseTime /3, TimeUnit.MILLISECONDS);}续期锁的lua脚本如下// 如果锁存在这个hashkey重新设置锁的过期时间if(redis.call(hexists, KEYS[1], ARGV[2])1)thenredis.call(pexpire, KEYS[1], ARGV[1]);return1;end;return0;到这里redisson实现分布式锁的源码解析就结束了。三、总结redisson的源码中大量使用了异步编程这导致阅读源码的难度系数较高这里我也只是大概整理了一下有问题的同学可以互相讨论一下或自行查阅源码。