Redisson分布式锁的实现原理及源码
文章图片

源码解析 简单的业务代码
主要负责源码入口, 分布式锁的使用主要有三个方法

1. RLock lock = redissonClient.getLock("hpc-lock")获取实现可重入分布式锁的类
2. lock.lock() 加锁
3. 【Redisson分布式锁的实现原理及源码】lock.unlock()解锁
   

   @GetMapping("/redis/lock") public ResResult testDistributedLock() { RLock lock = redissonClient.getLock("hpc-lock"); lock.lock(); try { System.out.println("加锁业务, xxx, xxx, xxxx"); } finally { lock.unlock(); } return new ResResult(true, ""); }

   
   

源码解析
获取实现可重入分布式锁的类 redissonClient.getLock("hpc-lock") ，该方法主要是获取实现了分布式可重入锁的类，进入getLock方法

@Override public RLock getLock(String name) { return new RedissonLock(commandExecutor, name); }

发现是初始化了RedissonLock类, 追到构造类方法

public RedissonLock(CommandAsyncExecutor commandExecutor, String name) { super(commandExecutor, name); this.commandExecutor = commandExecutor; // 通过下一行代码并进入追踪，发现默认线程的内部锁租用时间为默认的30s， // 也就是30s后自动释法锁 this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(); this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub(); }

加锁逻辑 lock.lock() 该方法主要功能是进行加锁，进入lock()方法，并向下追踪找到核心逻辑，找到方法org.redisson.RedissonLock#lock(long, java.util.concurrent.TimeUnit, boolean)并查看核心逻辑。
该核心逻辑主要有三个点：尝试加锁，未加锁成功的线程订阅Redis的消息， 未加锁成功的线程通过自旋获取锁
尝试加锁逻辑 首先看org.redisson.RedissonLock#lock(long, java.util.concurrent.TimeUnit, boolean)方法的尝试加锁部分，如下

long threadId = Thread.currentThread().getId(); // 获取锁 Long ttl = tryAcquire(-1, leaseTime, unit, threadId); // 获取锁成功，直接返回 if (ttl == null) { return; }

进入tryAcquire(-1, leaseTime, unit, threadId)方法并不断向下跟踪，找到核心逻辑org.redisson.RedissonLock#tryAcquireAsync方法体如下，该方法主要有几个逻辑：加锁，锁续命

private RFuture tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) { RFuture ttlRemainingFuture; // 获取锁 if (leaseTime != -1) { // 有锁失效时间 ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } else { // 采用默认锁失效时间 ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); } // 加锁后回调该方法 CompletionStage f = ttlRemainingFuture.thenApply(ttlRemaining -> { // 加锁成功 if (ttlRemaining == null) { if (leaseTime != -1) { internalLockLeaseTime = unit.toMillis(leaseTime); } else { // 如果没有传入锁失效时间，也就是在加锁时采用的是默认的锁失效时间 // 加锁成功后，进行锁续命 scheduleExpirationRenewal(threadId); } } return ttlRemaining; }); return new CompletableFutureWrapper<>(f); }

1. 首先，进入加锁的核心逻辑tryLockInnerAsync(waitTime, internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG), 方法体如下：
   

   RFuture tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) { // 对redis执行lua脚本 return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command, "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(getRawName()), unit.toMillis(leaseTime), getLockName(threadId)); }

   
   这里重要的是Lua脚本的逻辑，下面简单解释lua脚本的逻辑
   
   • 判断redis中是否存在具有getRawName()的键值的数据。（注意： 这里getRawName()所获取的值就是业务方法RLock lock = redissonClient.getLock("hpc-lock")中传入的参数hpc-lock）
   • 如果不存在键值，则保存该键值到redis中，并且该key对用的value值为1；同时该键值的失效时间设置为unit.toMillis(leaseTime)(这里的失效时间如果用户不传的话，一般采用默认的30s，这在之前的对redissonClient.getLock("hpc-lock")源码解析中已经分析过)，并返回null值。
   • 如果存在键值，则对该键值的value的值进行自增1，并且重新设置该键值的失效时间，失效时间设置为unit.toMillis(leaseTime)， 同时返回null值
   • 如果还有其它情况，则返回该键值的剩余过期时间，如果该键值不存在返回-2，如果该键值没有过期时间返回-1(这是Lua脚本中的return redis.call('pttl', KEYS[1]))
     注意： Lua脚本在Redis中是先天具有原子性的，只有Lua脚本执行完之后，Redis才会进行其它操作，因此不用担心Lua脚本的并发问题。
2. 当获取锁成功后，会进入回调方法，进行锁续命的逻辑，进入核心方法scheduleExpirationRenewal(threadId)中，方法体如下：
   

   protected void scheduleExpirationRenewal(long threadId) { ExpirationEntry entry = new ExpirationEntry(); ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry); if (oldEntry != null) { oldEntry.addThreadId(threadId); } else { entry.addThreadId(threadId); try { // 续期 renewExpiration(); } finally { // 当线程中断时，取消续期，这个下边分析，现不讨论 if (Thread.currentThread().isInterrupted()) { cancelExpirationRenewal(threadId); } } } }

   
   首先看续期的代码renewExpiration()，该方法内容较多，只看核心部分。其中internalLockLeaseTime的值在前文分析过，如果用户不传键值的有效期的话，默认为30s，通过下面代码可以看到该任务平均10s执行一次，也就是线程加锁后是10s续命一次。
   

   Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { //......// 进行续约 RFuture future = renewExpirationAsync(threadId); // 执行完续约后的回调 future.whenComplete((res, e) -> { if (e != null) { log.error("Can't update lock " + getRawName() + " expiration", e); EXPIRATION_RENEWAL_MAP.remove(getEntryName()); return; }if (res) { // 执行成功，回调自己 // reschedule itself renewExpiration(); } else { // 关闭续约 cancelExpirationRenewal(null); } }); } }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

   
   
   • 继续跟踪核心方法renewExpirationAsync(threadId)，查看该线程续约的核心逻辑
     

     protected RFuture renewExpirationAsync(long threadId) { return evalWriteAsync(getRawName(), 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(getRawName()), internalLockLeaseTime, getLockName(threadId)); }

     
     可以看到依然是执行Lua脚本，该脚本的逻辑是，如果该键值存在(仍在加锁/仍在执行锁内业务)时，重新设置Redis中该键值的失效时间internalLockLeaseTime(默认为30s，前文以分析)，并返回1；如果检测到Redis中不存在该键值，则直接返回0。
     
   • 看方法cancelExpirationRenewal(null)，关闭续约的方法
     

     protected void cancelExpirationRenewal(Long threadId) { ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (task == null) { return; }if (threadId != null) { task.removeThreadId(threadId); }if (threadId == null || task.hasNoThreads()) { Timeout timeout = task.getTimeout(); if (timeout != null) { // 如果有续约的定时任务，直接关闭 timeout.cancel(); } EXPIRATION_RENEWAL_MAP.remove(getEntryName()); } }

     
     

未加锁成功的线程订阅Redis消息 回到org.redisson.RedissonLock#lock(long, java.util.concurrent.TimeUnit, boolean)的方法中，通过上面尝试加锁逻辑的分析可以看到如果加锁成功后会直接返回，但未加锁成果会继续向下执行代码。
先看一下未加锁成功的线程订阅Redis消息的核心代码：

// 订阅消息 CompletableFuture future = subscribe(threadId); if (interruptibly) { commandExecutor.syncSubscriptionInterrupted(future); } else { commandExecutor.syncSubscription(future); }

继续进入代码subscribe(threadId)中，并继续追踪，查看如下核心逻辑，发现是未获取锁的线程通过信号量来监听接收信息

public CompletableFuture subscribe(String entryName, String channelName) { AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName)); CompletableFuture newPromise = new CompletableFuture<>(); int timeout = service.getConnectionManager().getConfig().getTimeout(); Timeout lockTimeout = service.getConnectionManager().newTimeout(t -> { newPromise.completeExceptionally(new RedisTimeoutException( "Unable to acquire subscription lock after " + timeout + "ms. " + "Increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters.")); }, timeout, TimeUnit.MILLISECONDS); semaphore.acquire(() -> { // ...... }); return newPromise; }

未加锁成功的线程通过自旋获取锁 回到org.redisson.RedissonLock#lock(long, java.util.concurrent.TimeUnit, boolean)的方法中，看一下通过自旋获取锁的代码如下

try { while (true) { // 尝试获取锁 ttl = tryAcquire(-1, leaseTime, unit, threadId); // 锁获取成功，直接跳出循环 if (ttl == null) { break; }// 等待信号量， if (ttl >= 0) { try { commandExecutor.getNow(future).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { if (interruptibly) { throw e; } commandExecutor.getNow(future).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } } else { if (interruptibly) { commandExecutor.getNow(future).getLatch().acquire(); } else { commandExecutor.getNow(future).getLatch().acquireUninterruptibly(); } } } } finally { // 取消订阅 unsubscribe(commandExecutor.getNow(future), threadId); }

释放锁的逻辑 业务代码中的lock.unlock()是用来解锁的代码，并向下跟踪核心方法代码如下

@Override public RFuture unlockAsync(long threadId) { // 释放锁代码 RFuture future = unlockInnerAsync(threadId); CompletionStage f = future.handle((opStatus, e) -> { // 取消到期续订，上文提到过，这里不再解释 cancelExpirationRenewal(threadId); if (e != null) { throw new CompletionException(e); } if (opStatus == null) { IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + id + " thread-id: " + threadId); throw new CompletionException(cause); }return null; }); return new CompletableFutureWrapper<>(f); }

上述方法有两个主要代码：释放锁，取消到期续订。取消到期续订的方法cancelExpirationRenewal(threadId)上文描述过。这里主要分析释放锁代码的方法，进入的方法unlockInnerAsync(threadId)中

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

这里主要还是执行的Lua脚本，Lua脚本的逻辑如下：

• 如果不存在该键值，直接返回null
• 该键值的value直接减1，再获取value的值，如果value的值仍大于0，则重新设置该键值的失效时间，然后返回0；如果value不大于0，则直接删除该键值，并发布订阅消息，并返回1
• 其它情况直接返回null

推荐阅读

• 

  北控水务为什么不涨 北控水务到底好不好
• 

  牵组词语 汉字牵组词语
• 

  月季花扦插的最佳时间 月季花扦插的最佳时间现在可以插吗?
• 

  2023重庆医科大学研究生成绩复查申请时间+方式
• 

  n95口罩能防尘吗 n95口罩可以用几次
• 

  摩托罗拉里程碑2刷机
• 

  如果你爱的人总是以怨报德,你会怎么样？
• 

  宝马525多少钱落地 宝马525价格多少钱
• 

  电商铺货是什么意思 电商货架什么意思，货架市场
• 

  胡萝卜怎样吃最有营养啊视频 胡萝卜怎样吃最有营养啊
• 

  巴戟天药材多少钱一斤，巴戟天药店多少钱一斤
• 

  吕锅上的污渍怎么去除啊
• 

  非主流图片唯美霸气 非主流相框素材
• 

  白金膏的功效与作用 皮白金膏的功效与作用?
• 

  虚拟机怎么开网页游戏模式，虚拟机怎样打开网页
• 

  闻鸡起舞的主人公是谁 囊萤映雪的主人公是谁
• 

  口红吃饭的时候怎么办
• 

  宝宝长期咳嗽易得肺炎
• 

  包贝尔大婚柳岩遭闹伴娘 柳岩伴娘事件是什么?
• 

  未婚生子罚款多少2022年 未婚生子罚款多少2022？

• 好家伙，分布式配置中心真好用
• 基于Redis分布式BitMap的应用
• 好家伙，分布式配置中心这种组件真的是神器
• 人人都能看懂系列:《分布式系统改造方案——老旧系统改造篇》
• 一文理解乐观锁与悲观锁
• 分布式文件上传导致服务假死了（）
• RC级别下MySQL死锁问题的解决
• vue实现滑动解锁功能
• vue滑动解锁组件使用方法详解
• 分布式|分布式ID生成器-订单号的生成(全局唯一id生成策略)