文章目录

Pre
用法
Redisson分布式锁实现原理
Redisson分布式锁源码分析
- redisson.getLock(lockKey) 的逻辑
- redissonLock.lock()的逻辑
- redissonLock.unlock();逻辑
总结

Pre

Redis进阶-细说分布式锁中我们梳理了使用Redis实现分布式锁的演进过程，并提出了目前最完善的解决方案：Redisson 实现分布式锁。

这里我们来分析下Redisson分布式锁实现原理及源码解析

用法

使用redisson实现分布式锁的操作步骤，三部曲

第一步：获取锁 RLock redissonLock = redisson.getLock(lockKey);
第二步：加锁，实现锁续命功能 redissonLock.lock();
第三步：释放锁 redissonLock.unlock();

Redisson分布式锁实现原理

熟悉了基本用法以后，我们来看下Redission实现分布式锁的原理，再理解了原理之后，后续梳理源码实现就更加得心应手了。

Redisson分布式锁源码分析

流程图如下

重点主要是依赖lua脚本的原子性，实现加锁和释放锁的功能

redisson.getLock(lockKey) 的逻辑

    @Overridepublic RLock getLock(String name) {return new RedissonLock(connectionManager.getCommandExecutor(), name);}

实例化RedissonLock，我们看下RedissonLock的构造函数

    public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {super(commandExecutor, name);this.commandExecutor = commandExecutor;this.id = commandExecutor.getConnectionManager().getId();this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();}

super(commandExecutor, name); 父类name赋值，后续通过getName()获取
commandExecutor：执行lua脚本的executor
id 是个UUID，后面被用来当做和threadId组成 value值，用作判断加锁和释放锁是否是同一个线程的校验。
internalLockLeaseTime : 取自 Config#lockWatchdogTimeout，默认30秒，这个参数还有另外一个作用，锁续命的执行周期 internalLockLeaseTime/3 = 10秒

redissonLock.lock()的逻辑

主要是实现加锁和锁的续命

 redissonLock.lock();

看看都干了啥

  @Overridepublic void lock() {try {lockInterruptibly();} catch (InterruptedException e) {Thread.currentThread().interrupt();}}

继续看 lockInterruptibly

 @Overridepublic void lockInterruptibly() throws InterruptedException {lockInterruptibly(-1, null);}

继续看 lockInterruptibly(-1, null);

@Overridepublic void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {// 获取当前线程IDlong threadId = Thread.currentThread().getId();// 尝试获取锁的剩余时间 Long ttl = tryAcquire(leaseTime, unit, threadId);// lock acquired  ttl为空，说明没有线程持有该锁，直接返回 让当前线程加锁成功 if (ttl == null) {return;}RFuture<RedissonLockEntry> future = subscribe(threadId);commandExecutor.syncSubscription(future);// 死循环  try {while (true) {// 再此尝试获取锁的剩余时间 ，如果为null, 跳出循环ttl = tryAcquire(leaseTime, unit, threadId);// lock acquiredif (ttl == null) {break;}// waiting for message   如果ttl >=0 说明 有其他线程持有该锁if (ttl >= 0) {// 获取信号量，尝试加锁，设置最大等待市场为ttlgetEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {// 如果ttl小于0 （-1 ，-2 ） 说明已经过期，直接获取getEntry(threadId).getLatch().acquire();}}} finally {unsubscribe(future, threadId);}
//        get(lockAsync(leaseTime, unit));}

大流程已经梳理完了，我们看下 Long ttl = tryAcquire(leaseTime, unit, threadId);

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

继续看下

tryAcquireAsync(leaseTime, unit, threadId)

 private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {if (leaseTime != -1) {return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);}// 刚开始  leaseTime 传入的是 -1 ，所以走这个分支// 1）尝试加锁  待会细看 先把主要的逻辑梳理完RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);// 2) 注册监听事件ttlRemainingFuture.addListener(new FutureListener<Long>() {@Overridepublic void operationComplete(Future<Long> future) throws Exception {if (!future.isSuccess()) {return;}Long ttlRemaining = future.getNow();// lock acquiredif (ttlRemaining == null) {// 3）获取锁成功的话，给锁延长过期时间 scheduleExpirationRenewal(threadId);}}});return ttlRemainingFuture;}

继续看

 // 1）尝试加锁  待会细看 先把主要的逻辑梳理完RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);

看实现

    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {internalLockLeaseTime = unit.toMillis(leaseTime);return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hset', 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));}

lua 脚本

KEYS[1] ---------> getName()
ARGV[1] ---------> internalLockLeaseTime
ARGV[2] ---------> getLockName(threadId) 实现如下

  String getLockName(long threadId) {return id + ":" + threadId;}

这个id就是自开始实例化RedissonLock的id ,是个UUID

我们来解释下这段lua脚本

 // 如果 lockKey不存在 ，设置 使用hset设置 lockKey ，field为 uuid:threadId ,value为1 ,并设置过期时间//就是这个命令 //127.0.0.1:6379> hset lockkey  uuid:threadId 1//(integer) 1//127.0.0.1:6379> PEXPIRE lockkey internalLockLeaseTime"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hset', KEYS[1], ARGV[2], 1); " +"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +// 如果 lockKey 存在和 filed 和 当前线程的uuid:threadId相同  key 加1 ，执行多少次 就加多次  设置过期时间  其实就是如下命令//127.0.0.1:6379> HEXISTS lockkey uuid:threadId//(integer) 1//127.0.0.1:6379> PEXPIRE lockkey  internalLockLeaseTime"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; " +// 最后返回 lockkey的 pttl "return redis.call('pttl', KEYS[1]);"

那继续监听时间中的 scheduleExpirationRenewal(threadId); 逻辑

 private void scheduleExpirationRenewal(final long threadId) {if (expirationRenewalMap.containsKey(getEntryName())) {return;}Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {// 重点是run方法 @Overridepublic void run(Timeout timeout) throws Exception {// 又是lua脚本  判断是否存在，存在就调用pexpire RFuture<Boolean> future = commandExecutor.evalWriteAsync(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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));// 监听事件中又 调用了自己  scheduleExpirationRenewalfuture.addListener(new FutureListener<Boolean>() {@Overridepublic void operationComplete(Future<Boolean> future) throws Exception {expirationRenewalMap.remove(getEntryName());if (!future.isSuccess()) {log.error("Can't update lock " + getName() + " expiration", future.cause());return;}if (future.getNow()) {// reschedule itselfscheduleExpirationRenewal(threadId);}}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {task.cancel();}}

redissonLock.unlock();逻辑

  @Overridepublic void unlock() {Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));if (opStatus == null) {throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "+ id + " thread-id: " + Thread.currentThread().getId());}if (opStatus) {cancelExpirationRenewal();}}

重点看 unlockInnerAsync(Thread.currentThread().getId())

 protected RFuture<Boolean> unlockInnerAsync(long threadId) {return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; " +"end;" +"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.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));}

又是lua脚本，核心就是把value减到为0 ，删除key

KEYS[1] ---------> getName()

KEYS[2] ---------> getChannelName()
ARGV[1] ---------> LockPubSub.unlockMessage
ARGV[2] ---------> internalLockLeaseTime
ARGV[2] ---------> getLockName(threadId)

总结

需要用到续锁功能时，一要记住不要设置锁的过期时间，可以设置成-1.

一旦设了时间，RedissonLock就会认为你需要自己控制锁时间，而放弃执行续锁逻辑。
查看源码，续锁逻辑需要起定时器。所以要注意这点，并不是所有分布式场景都需要续锁逻辑的。当我们很难判断业务逻辑的执行时间时，不妨开启续锁。

至此，原理和源码我们粗略的梳理完了，梳理了主要的核心流程，主要是依靠lua脚本，代码写的还是非常优秀的，向开源学习！！！

Redis进阶- Redisson分布式锁实现原理及源码解析相关推荐

【精通内核】Linux内核自旋锁实现原理与源码解析
前言
Redis进阶-细说分布式锁
文章目录 Pre 引分布式锁演进 V1 分布式锁演进 V2 分布式锁演进 V3 分布式锁演进 V4 分布式锁演进 V5 终极版-分布式锁演进(Redisson ) V6 Code Redisson分 ...
Redis：Redisson分布式锁的使用（推荐使用）
Redis:Redisson分布式锁的使用(生产环境下)(推荐使用) 关键词基于NIO的Netty框架,生产环境使用分布式锁 redisson加锁:lua脚本加锁(其他客户端自旋) 自动延时机制:启 ...
17、Redis、Zk分布式锁实现原理
我们在编程有很多场景使用本地锁和分布式锁,但是是否考虑这些锁的原理是什么?本篇讨论下实现分布式锁的常见办法及他们实现原理. 一.使用锁的原则使用本地锁和分布式锁是为了解决并发导致脏数据的场景,使用锁 ...
redis ，redisson 分布式锁深入剖析
目录为什么要用分布式锁? 分布式锁所遵循的原则? redis 分布式锁 redis 原始分布式锁实现加锁释放锁 redis 分布式锁存在的问题 redisson 实现分布式锁 redisson ...
Dubbo 实现原理与源码解析系列 —— 精品合集
摘要: 原创出处 http://www.iocoder.cn/Dubbo/good-collection/ 「芋道源码」欢迎转载,保留摘要,谢谢! 1.[芋艿]精尽 Dubbo 原理与源码专栏 2.[ ...
【特征匹配】ORB原理与源码解析
相关 : Fast原理与源码解析 Brief描述子原理与源码解析 Harris原理与源码解析 http://blog.csdn.net/luoshixian099/article/details/48 ...
PCA-SIFT原理及源码解析
相关: SIFT原理与源码解析 SURF原理与源码解析 ORB原理与源码解析 FAST原理与源码解析 BRIEF描述子原理与源码解析 Harris原理与源码解析转载请注明出处:http://blog ...
Spring Boot 核心原理与源码解析 - 目录
准备重新写 SpringBoot 配置文件解析原理 , 先在这里把要写的内容记下来 Spring Boot 核心原理与源码解析 - 目录 1\何时解析\如何解析 application.propert ...

Redis进阶- Redisson分布式锁实现原理及源码解析