Redisson公平锁的源码解读分享
作者：王谷雨 发布时间：2022-12-13 01:20:40
标签：Redisson,公平锁
前言
我在上一篇文章聊了Redisson的分布式锁，这次继续来聊聊Redisson的公平锁。下面是官方原话：
它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待5秒后继续下一个线程，也就是说如果前面有5个线程都处于等待状态，那么后面的线程会等待至少25秒。
源码版本：3.17.7
这是我 fork 的分支，添加了自己理解的中文注释：https://github.com/xiaoguyu/redisson
公平锁
先上官方例子：
RLock fairLock = redisson.getFairLock("anyLock");
// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);
...
fairLock.unlock();
因为在Redisson中，公平锁和普通可重入锁的逻辑大体上一样，我在上一篇文章都介绍了，这里就不再赘述。下面开始介绍合理逻辑。
加锁
加锁的 lua 脚本在 RedissonFairLock#tryLockInnerAsync方法中
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
long wait = threadWaitTime;
if (waitTime > 0) {
wait = unit.toMillis(waitTime);
}
long currentTime = System.currentTimeMillis();
if (command == RedisCommands.EVAL_NULL_BOOLEAN) {
......
}
if (command == RedisCommands.EVAL_LONG) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
// remove stale threads
"while true do " + // list为空，证明没有人排队，退出循环
"local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +
"if firstThreadId2 == false then " +
"break;" +
"end;" +
// 能到这里，证明有人排队，拿出在排队的第一个人的超时时间，如果超时了，则移除相应数据
"local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +
"if timeout <= tonumber(ARGV[4]) then " +
// remove the item from the queue and timeout set
// NOTE we do not alter any other timeout
"redis.call('zrem', KEYS[3], firstThreadId2);" +
"redis.call('lpop', KEYS[2]);" +
"else " +
"break;" +
"end;" +
"end;" +
// check if the lock can be acquired now
// 检查是否可以获取锁。如果hash和list都不存在，或者线程队列的第一个是当前线程，则可以获取锁
"if (redis.call('exists', KEYS[1]) == 0) " +
"and ((redis.call('exists', KEYS[2]) == 0) " +
"or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " +
// remove this thread from the queue and timeout set
// 都获取锁了，当然要从线程队列和时间队列中移除
"redis.call('lpop', KEYS[2]);" +
"redis.call('zrem', KEYS[3], ARGV[2]);" +
// decrease timeouts for all waiting in the queue
// 刷新时间集合中的时间
"local keys = redis.call('zrange', KEYS[3], 0, -1);" +
"for i = 1, #keys, 1 do " +
"redis.call('zincrby', KEYS[3], -tonumber(ARGV[3]), keys[i]);" +
"end;" +
// acquire the lock and set the TTL for the lease
// 和公平锁的设置一样，值加1并且设置过期时间
"redis.call('hset', KEYS[1], ARGV[2], 1);" +
"redis.call('pexpire', KEYS[1], ARGV[1]);" +
"return nil;" +
"end;" +
// check if the lock is already held, and this is a re-entry
// 能到这里，证明前面拿不到锁，但是也要做可重入锁的处理
"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;" +
// the lock cannot be acquired
// check if the thread is already in the queue
// 时间集合中有值，证明线程已经在队列中，不需要往后执行逻辑了
"local timeout = redis.call('zscore', KEYS[3], ARGV[2]);" +
"if timeout ~= false then " +
// the real timeout is the timeout of the prior thread
// in the queue, but this is approximately correct, and
// avoids having to traverse the queue
// 因为下面的timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4])
// 所以这里的ttl = timeout - tonumber(ARGV[3]) - tonumber(ARGV[4])
"return timeout - tonumber(ARGV[3]) - tonumber(ARGV[4]);" +
"end;" +
// add the thread to the queue at the end, and set its timeout in the timeout set to the timeout of
// the prior thread in the queue (or the timeout of the lock if the queue is empty) plus the
// threadWaitTime
"local lastThreadId = redis.call('lindex', KEYS[2], -1);" +
"local ttl;" +
// 如果最后一个线程不是当前线程，则从时间集合取出（举例：线程1/2/3按顺序获取锁，此时pttl得到的是线程1的锁过期时间，zscore拿到的是线程2的锁的过期时间，此时线程3应该以线程2的为准）
"if lastThreadId ~= false and lastThreadId ~= ARGV[2] then " +
"ttl = tonumber(redis.call('zscore', KEYS[3], lastThreadId)) - tonumber(ARGV[4]);" +
"else " +
// 否则直接获取锁的存活时间
"ttl = redis.call('pttl', KEYS[1]);" +
"end;" +
// 过期时间 = 锁存活时间 + 等待时间 + 当前时间戳
"local timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4]);" +
// 如果添加到时间集合成功，则同时添加线程集合
"if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " +
"redis.call('rpush', KEYS[2], ARGV[2]);" +
"end;" +
"return ttl;",
Arrays.asList(getRawName(), threadsQueueName, timeoutSetName),
unit.toMillis(leaseTime), getLockName(threadId), wait, currentTime);
}
throw new IllegalArgumentException();
}
公平锁总共用了Redis的三种数据类型，对应着 lua 脚本里面的keys1、2、3的参数：
KEYS[1]
锁的名字，使用 Hash 数据类型，是可重入锁的基础，结构为 {”threadId1”: 1, “thread2”: 1}，key为线程id，value是锁的次数
KEYS[2]
线程队列的名字，使用 List 数据类型，结构为 [ “threadId1”, “threadId2” ]，按顺序存放需要获取锁的线程的id
KEYS[3]
时间队列的名字，使用 sorted set 数据类型，结构为 {”threadId2”:123, “threadId1”:190}，key为线程id，value为获取锁的超时时间戳
我下面会用 锁、线程队列、时间队列 来表示这3个数据结构，需要注意下我的表述。
同样的，介绍下参数：
ARGV[1]：leaseTime 锁的持有时间
ARGV[2]：线程id（描述不太准确，暂时按这样理解）
ARGV[3]：waitTime 尝试获取锁的最大等待时间
ARGV[4]：currentTime 当前时间戳
接下来，我们一段一段分析 lua 脚本，首先看最开始的 while 循环
"while true do " + // list为空，证明没有人排队，退出循环
"local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +
"if firstThreadId2 == false then " +
"break;" +
"end;" +
// 能到这里，证明有人排队，拿出在排队的第一个人的超时时间，如果超时了，则移除相应数据
"local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +
"if timeout <= tonumber(ARGV[4]) then " +
// 从时间队列和线程队列中移除
"redis.call('zrem', KEYS[3], firstThreadId2);" +
"redis.call('lpop', KEYS[2]);" +
"else " +
"break;" +
"end;" +
"end;" +
具体的逻辑我在注释中写的很清楚了，看的时候记住 KEYS[2]、KEYS[3] 对应着线程队列和时间队列接口。主要注意的是，线程队列只有当一个线程持有锁，另一个线程获取不到锁时，才会有值（前面有人才排队，没人排什么队）。接着看第二段
// 检查是否可以获取锁。当锁不存在，并且线程队列不存在或者线程队列第一位是当前线程，则可以获取锁
"if (redis.call('exists', KEYS[1]) == 0) " +
"and ((redis.call('exists', KEYS[2]) == 0) or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " +
// remove this thread from the queue and timeout set
// 都获取锁了，当然要从线程队列和时间队列中移除
"redis.call('lpop', KEYS[2]);" +
"redis.call('zrem', KEYS[3], ARGV[2]);" +
// decrease timeouts for all waiting in the queue
// 刷新时间队列中的时间
"local keys = redis.call('zrange', KEYS[3], 0, -1);" +
"for i = 1, #keys, 1 do " +
"redis.call('zincrby', KEYS[3], -tonumber(ARGV[3]), keys[i]);" +
"end;" +
// acquire the lock and set the TTL for the lease
// 和公平锁的设置一样，值加1并且设置过期时间
"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;" +
这是可重入锁的处理，继续下一段
// 时间队列中有值，证明线程已经在队列中，不需要往后执行逻辑了
"local timeout = redis.call('zscore', KEYS[3], ARGV[2]);" +
"if timeout ~= false then " +
// the real timeout is the timeout of the prior thread
// in the queue, but this is approximately correct, and
// avoids having to traverse the queue
// 因为下面的timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4])
// 所以这里的ttl = timeout - tonumber(ARGV[3]) - tonumber(ARGV[4])
"return timeout - tonumber(ARGV[3]) - tonumber(ARGV[4]);" +
"end;" +
举例子：线程1持有锁，线程2尝试第一次获取锁（不进入这段if），线程2第二次获取锁（进入了这段if）。继续下一段
"local lastThreadId = redis.call('lindex', KEYS[2], -1);" +
"local ttl;" +
// 如果最后一个线程不是当前线程，则从时间集合取出（举例：线程1/2/3按顺序获取锁，此时pttl得到的是线程1的锁过期时间，zscore拿到的是线程2的锁的过期时间，此时线程3应该以线程2的为准）
"if lastThreadId ~= false and lastThreadId ~= ARGV[2] then " +
"ttl = tonumber(redis.call('zscore', KEYS[3], lastThreadId)) - tonumber(ARGV[4]);" +
"else " +
// 否则直接获取锁的存活时间
"ttl = redis.call('pttl', KEYS[1]);" +
"end;" +
// 过期时间 = 锁存活时间 + 等待时间 + 当前时间戳
"local timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4]);" +
// 如果添加到时间集合成功，则同时添加线程集合
"if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " +
"redis.call('rpush', KEYS[2], ARGV[2]);" +
"end;" +
"return ttl;",
ttl 这段的获取逻辑，翻译翻译就是，如果前面有人排队，就以前面的超时时间为准，如果没人排队，就拿锁的超时时间。获取到 ttl ，就对添加到线程集合和时间集合。
来源：https://www.cnblogs.com/konghuanxi/p/16869998.html
0
投稿猜你喜欢
- 与任何程序设计语言一样，Java使用条件语句和循环结构确定控制流。本文将简单讲解条件、循环和switch。一、块作用域块（block），即复
本文主要介绍了Spring Security OAuth2 实现登录互踢的示例代码，分享给大家，具体如下：背景说明一个账号只能一处登录，类似- JVM自带的类加载器：其关系如下：其中，类加载器在加载类的时候是使用了所谓的“父委托”机制。其中，除了根类加载器以外，其他的类加载器都有且只
- Map集合和Collection集合的区别Map集合是有Key和Value的，Collection集合是只有Value。Collection
- 现象在日志配置文件 logback-spring.xml 中，无论怎么修改级别，mybatis 的 sql 日志都会打印出来。原因在 app
- 方法一、利用控件或窗体的Paint事件中的PainEventArgs在窗体或控件的Paint事件中接收对图形对象的引用，作为PaintEve
- 高快省的排序算法有没有既不浪费空间又可以快一点的排序算法呢？那就是“快速排序”啦！光听这个名字是不是就觉得很高端呢。假设我们现在对“6 1
- 简单看一下描述，例子最重要。1、getPath()：返回定义时的路径，（就是你写什么路径，他就返回什么路径）2、getAbsolutePat
- Java 实现汉字转换为拼音转换类public class PINYINChinese { private static int
- Maven 翻译为"专家"、"内行"，是 Apache 下的一个纯 Java 开发的开源项目。基于项
- 以下实例演示了如何使用 equals ()方法来判断数组是否相等：import java.util.Arrays; public class
- 至少有K个重复字符的最长子串给你一个字符串 s 和一个整数 k ，请你找出 s 中的最长子串， 要求该子串中的每一字符出现次数都不
- AsyncTask什么是AsyncTaskAsyncTask是一个轻量级的异步任务类，它可以在线程池中执行后台任务，然后把执行的进度和结果传
- 1、synchronized的作用为了避免临界区的竞态条件发生，有多种手段可以达到目的。阻塞式的解决方案：synchronized，Lock
- import java.awt.AlphaComposite;import java.awt.Color;import java.awt.F
- 有一天，你写了好多好多带“形参”的构造函数（就是“方法”，同义），而且需要向这些构造函数里传递同样的“实参”，然后你就憨憨地一个一个函数的调
- 1.定义Token的注解，需要Token校验的接口，方法上加上此注解import java.lang.annotation.ElementT
- 本文首先将会回顾Spring 5之前的SpringMVC异常处理机制，然后主要讲解Spring Boot 2 Webflux的全局异常处理机
- 一、maven引入依赖，数据库驱动根据项目需求自行引入<!-- https://mvnrepository.com/artifact/
- 上篇给大家介绍了Spring Boot启动过程完全解析(一)，大家可以点击参考下该说refreshContext(context)了，首先是
