【源码阅读】Redisson lock源码

愛在花開的季節  金牌会员 | 2024-8-3 14:25:04 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 551|帖子 551|积分 1653

目录
底层原理
加锁机制
锁互斥机制
可重入锁机制
总结 
Redisson 加锁非常简单，还支持 redis 单实例、redis 哨兵、redis cluster、redis master-slave 等各种部署架构
  1. RLock lock = redisson.getLock("cyk-test");
  2. lock.lock();
  3. lock.unlock();
复制代码
底层原理



加锁机制

废话不多说，直接看源码，下面的代码先不看，先看 tryAcquire 是怎样获取锁的
  1. private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
  2.         long threadId = Thread.currentThread().getId();
  3.         Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
  4.         // lock acquired
  5.         if (ttl == null) {
  6.             return;
  7.         }
  9.         CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
  10.         pubSub.timeout(future);
  11.         RedissonLockEntry entry;
  12.         if (interruptibly) {
  13.             entry = commandExecutor.getInterrupted(future);
  14.         } else {
  15.             entry = commandExecutor.get(future);
  16.         }
  17.         ...
  18. }
复制代码
查看 tryAcquire 方法，点进去看发现调用了 tryAcquireAsync0 方法，这里 RFuture 继承自 java.util.concurrent.Future，表现这是一个异步的使命，get 方法会同步获取效果
  1. private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
  2.     return get(tryAcquireAsync0(waitTime, leaseTime, unit, threadId));
  3. }
  5. private RFuture<Long> tryAcquireAsync0(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
  6.     return getServiceManager().execute(() -> tryAcquireAsync(waitTime, leaseTime, unit, threadId));
  7. }
复制代码
查看 tryAcquireAsync 方法
  1.     private RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
  2.         RFuture<Long> ttlRemainingFuture;
  3.         if (leaseTime > 0) {
  4.             ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
  5.         } else {
  6.             ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
  7.                     TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
  8.         }
  9.         ...
  10.     }
复制代码
查看 tryLockInnerAsync 方法
  1.     <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
  2.         return commandExecutor.syncedEval(getRawName(), LongCodec.INSTANCE, command,
  3.                 "if ((redis.call('exists', KEYS[1]) == 0) " +
  4.                             "or (redis.call('hexists', KEYS[1], ARGV[2]) == 1)) then " +
  5.                         "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
  6.                         "redis.call('pexpire', KEYS[1], ARGV[1]); " +
  7.                         "return nil; " +
  8.                     "end; " +
  9.                     "return redis.call('pttl', KEYS[1]);",
  10.                 Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
  11.     }
复制代码
这是一段加锁的 lua 脚本，用于包管原子性，参数解释如下：


  • KEYS[1] 表现加锁的 key
  • ARGV[1] 表现锁 key 的默认超时时间
  • ARGV[2] 表现加锁的客户端 ID，由 UUID:线程 ID 构成
客户端在第一次加锁完成，会设置一个 key 为客户端 ID，value 为加锁次数的 hash 数据结构：


现在知道了内部方法的逻辑，往回倒一步，重点看我加在代码中的注释
  1.     private RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
  2.         RFuture<Long> ttlRemainingFuture;
  3.         if (leaseTime > 0) {
  4.             ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
  5.         } else {
  6.             ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
  7.                     TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
  8.         }
  10.         // 这里是定义了加锁Lua脚本的异步任务，通过CompletableFuture编排
  11.         CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);
  12.         ttlRemainingFuture = new CompletableFutureWrapper<>(s);
  14.         // 这个f依赖ttlRemainingFuture的结果，如果入参的leaseTime<=0会触发看门狗机制
  15.         // 否则按照设置的过期时间来过期
  16.         CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
  17.             // lock acquired
  18.             if (ttlRemaining == null) {
  19.                 if (leaseTime > 0) {
  20.                     internalLockLeaseTime = unit.toMillis(leaseTime);
  21.                 } else {
  22.                     scheduleExpirationRenewal(threadId);
  23.                 }
  24.             }
  25.             return ttlRemaining;
  26.         });
  28.         // 返回编排好的CompletableFuture
  29.         return new CompletableFutureWrapper<>(f);
  30.     }
复制代码
把 RFuture 返回以后，就到了 get 方法阻塞获取方法这里
  1. private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
  2.     return get(tryAcquireAsync0(waitTime, leaseTime, unit, threadId));
  3. }
  5. private RFuture<Long> tryAcquireAsync0(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
  6.     return getServiceManager().execute(() -> tryAcquireAsync(waitTime, leaseTime, unit, threadId));
  7. }
复制代码
最后回到了这里
  1.     private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
  2.         long threadId = Thread.currentThread().getId();
  3.         Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
  4.         // lock acquired
  5.         if (ttl == null) {
  6.             return;
  7.         }
  9.         // 这里的 subscribe 就是 Redis 订阅解锁的 lua 脚本中的 publish
  10.         CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
  11.         pubSub.timeout(future);
  12.         RedissonLockEntry entry;
  13.         if (interruptibly) {
  14.             entry = commandExecutor.getInterrupted(future);
  15.         } else {
  16.             entry = commandExecutor.get(future);
  17.         }
  18.         ...
  19.     }
复制代码
接着看下面循环获取锁的逻辑
  1.         try {
  2.            while (true) {
  3.                 // 自旋尝试获取锁
  4.                ttl = tryAcquire(-1, leaseTime, unit, threadId);
  5.                // 看Lua脚本，ttl为null说明锁获取到了
  6.                if (ttl == null) {
  7.                    break;
  8.                }
  9. ​
  10.                // waiting for message
  11.                if (ttl >= 0) {
  12.                    try {
  13.                        // 注意这里的tryAcquire和之前的tryAcquire不是同一个东西，这里是信号量的tryAcquire
  14.                        // entry.getLatch()这里返回的是信号量，释放锁的代码会释放一个许可
  15.                         // 如果没有可用的许可，会一直休眠直到超时时间 ttl ms
  16.                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
  17.                    } catch (InterruptedException e) {
  18.                        if (interruptibly) {
  19.                            throw e;
  20.                        }
  21.                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
  22.                    }
  23.                } else {
  24.                    // 当 key 存在但没有设置剩余生存时间时，pttl返回 -1，会走到这个逻辑
  25.                     // 我感觉正常流程走不到这个逻辑，因为当前线程无非是看到锁存在或者不存在
  26.                     // 看到锁不存在等于加锁成功了，因为Lua脚本是原子性的
  27.                     // 看到锁存在，默认也给了超时时间
  28.                    // 这里就没有设置超时时间，一直等释放锁的许可
  29.                    if (interruptibly) {
  30.                        entry.getLatch().acquire();
  31.                    } else {
  32.                        entry.getLatch().acquireUninterruptibly();
  33.                    }
  34.                }
  35.            }
  36.        } finally {
  37.            unsubscribe(entry, threadId);
  38.        }
复制代码
这里计划的奇妙之处就在于利用了消息订阅、信号量的机制，它不是无休止的盲等机制，也避免了不断的重试，而是检测到锁被释放才去尝试重新获取，这对 CPU 十分的友爱
锁互斥机制

此时如果客户端 2 来尝试加锁，同样走进 RedissonLock#lock 方法，会咋样呢？第一个 if 判断会实行 exists myLock，发现 myLock 这个锁 key 已经存在了。接着第二个 if 判断，判断一下，myLock 锁 key 的 hash 数据结构中，对应客户端 2 的 ID 的 key 的 value 为 1，也没有。最终会获取到 pttl myLock 返回的锁 key 的剩余生存时间，进入 while 循环，不停的尝试加锁
可重入锁机制

那如果客户端1都已经持有了这把锁了，效果可重入的加锁会怎么样呢？


此时会实行可重入加锁的逻辑，走第二个 if 逻辑，对客户端 1 的加锁次数累加 1，此时 myLock 数据结构变为下面这样：


总结 




免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

愛在花開的季節

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表