Redisson分布式延时队列 RedissonDelayedQueue

自己实现java.util.concurrent.Delayed这个接口,重写getDelay(计算剩余时间)方法,重写compareTo(比较哪个任务更早过期，计算优先级)方法。

调用add()方法添加延迟任务，add()方法里面会调用offer()方法,offer()方法会调用q.offer(),q是PriorityQueue优先级队列，所以数据最终是存放在PriorityQueue优先级队列里面的。offer()方法会先加锁，同一时间只能有一个线程添加数据。 q.offer()方法内部会调用我们自己重写的compareTo()方法进行重新排序。

add()方法在调用q.offer()方法添加完成并排序成功之后，add()方法会将排序后的第一个延迟任务拿出来跟我们当前的延迟任务进行比较，如果是同一个延迟任务说明最先到期的延迟任务就是我们刚添加进去的这个延迟任务并且同时也说明之前队列里面的优先级有变化，此时需要调用signal()方法通知(唤醒)等待获取延迟任务的线程，延迟任务的优先级发生变化了，可以来取新的延迟任务了。 至此，我们往延迟队列里面添加延迟任务的工作就已经彻底做完了。

从延迟队列DelayQueue里面取出延迟队列任务，会调用DelayQueue.take()方法，take()方法里面是一个死循环。循环里面会调用q.peek()方法，从优先级队列里面取出排在第一位的延迟任务，如果取不到延迟任务，说明没有延迟任务，调用await()方法，让线程永远阻塞下去。如果取到延迟任务，调用延迟任务对象的getDelay()方法获取延迟任务的剩余时间，如果剩余时间小于等于0，说明延期已经过期了，调用p.poll()方法，将延迟任务return出去。如果剩余时间大于0，调用awaitNanos(延迟任务剩余时间)方法,让线程阻塞，阻塞时间为延迟任务的剩余时间。到时间之后，由于take方法是死循环，代码会重新调用p.peek取出元素并判断剩余时间是否小于等于0。take()方法一样也会加锁，同一个时间只能有一个线程从里面取出数据。

Java原生自带的DelayQueue的代码示例如下：

源码分析参考这篇文章《DelayQueue延迟队列》

pom.xml

SpringBoot的application.yml配置文件

Redisson配置类:RedissonConfig.java

使用Redisson添加任务到延迟队列里面

Controller代码

**offer()**方法是往Redisson的RDelayedQueue延迟队列里面添加延迟任务的，所以先看这个方法。

在IDEA里面按住鼠标左键点击redissonClient.getDelayedQueue(blockingDeque)方法，很容易知道RDelayedQueue的实现类为:org.redisson.RedissonDelayedQueue.java。那么offer方法的源码也在org.redisson.RedissonDelayedQueue.java里面，如下截图：

offer方法的源码

offer方法的源码

其实，offer方法的核心原理很简单。源码里面的Lua脚本我们不用管，我们只看我们能看懂的JAVA代码和Redis命令。

首先，用户传进来的延迟时间必须大于0，小于0抛出异常代码结束。

将用户传进来的时间转换为毫秒，并加上系统当前时间，计算出来的就是过期时间。到了过期时间消费者就可以把该任务取出来消费了。

最为核心的来了，第一条Redis命令：zadd KEYS[2], ARGV[1], value。注意：Lua的数组下标是从1开始的，跟JAVA的不一样，不是从0开始的。所以这条命令翻译过来就是:zadd timeoutSetName timeout value，使用zadd命令将用户的延迟任务存放在Redis的SortedSet数据结构里面，score的值为延迟任务的过期时间。Redis的SortedSet数据结构是一个自动排序的Set集合，所以使用zadd命令将延迟任务放进Redis的SortedSet数据结构之后，Redis的SortedSet会自动按照score(延迟任务的过期时间)从小到大自动排序，最先过期的任务排在最上面。

第二条Redis命令：rpush KEYS[3], value。这条命令翻译过来就是:rpush queueName value，使用Redis的rpush命令，将延迟任务存放在Redis的List数据结构里面。注意：这个list里面只存了具体的任务，没有存任务的过期时间。

第三条Redis命令：zrange KEYS[2], 0, 0。这条命令翻译过来就是:zrange timeoutSetName, 0, 0。从Redis的SortedSet数据结构取出第一条也就是最先过期的延迟任务。

if v[1] == value then的意思是：将排序后的第一个延迟任务拿出来跟我们当前的延迟任务进行比较，如果是同一个延迟任务说明最先到期的延迟任务就是我们刚添加进去的这个延迟任务并且同时也说明之前队列里面的优先级有变化，此时使用Redis的发布订阅命令：publish KEYS[4], ARGV[1] => publish channelName timeout，将延迟任务的过期时间timeout发布出去，通知订阅者，延迟任务的优先级发生变化了，可以来取新的延迟任务了。 至此，我们往延迟队列里面添加延迟任务的工作就已经彻底做完了。

其实，Redisson的这个流程跟Java原生自带的DelayQueue的offer方法流程是一样的。

**take()**方法是从Redisson的RDelayedQueue延迟队列里面取出延迟任务的。

take()这个方法的核心原理非常简单，核心代码就是使用Redis的BLPOP命令，从Redis的List数据结构里面取数据，取不到就阻塞，一直等到有数据进来。源码位置:org.redisson.RedissonBlockingDeque，源码如下截图：

org.redisson.RedissonBlockingDeque源码

org.redisson.RedissonBlockingQueue源码

org.redisson.client.protocol.RedisCommands源码

所以blockingDeque.take()方法的核心代码就是一条Redis命令:blpop 队列名字 0，队列名字就是你在 redissonClient.getBlockingDeque("orderQueue")这个方法里面传入的名字:orderQueue。发送给Redis执行的命令就是这个:blpop orderQueue 0,0代表不限时，一直阻塞下去。Redisson会把所有已经过期的任务，都存放在这个List里面，所以只有这个List里面有数据，就代表这个数据已经过期了，消费者可以消费了。Redisson往这个List里面放的时候使用的rpush命令，rpush命令的意思是往List的右边放。比如有A和B俩个任务，先放A,再放B，List里面的数据为:AB。但是注意先放进去的肯定是最先过期的，所以我们消费的时候要先消费A，再消费B。 而blpop这个命令就是从List的左边开始消费的。一唱一和，非常完美。

还记得下面这段代码吗？

注意:上面这俩行代码，无论你是在调用offer方法存放数据还是调用take()方法取出数据，这俩行代码都是必不可少的。

过期的数据就是在这俩行代码里面放进去的，核心代码在这里：org.redisson.Redisson

源码如下截图：

redissonClient.getDelayedQueue(blockingDeque)方法源码

redissonClient.getDelayedQueue(blockingDeque)方法调用了RedissonDelayedQueue类的构造方法，过期的数据就是在这个构造方法里面放进去的。

org.redisson.RedissonDelayedQueue构造方法的源码

我们先看上图里面pushTaskAsync方法里面的Redis命令。

第一条Redis命令:zrangebyscore KEYS[2] 0 ARGV[1] limit 0 ARGV[2],翻译过来为：zrangebyscore timeoutSetName 0 System.currentTimeMillis() limit 0 100。意思就是从timeoutSetName这个SortedSet数据结构里面取出score的值在0到系统当前时间的所有延迟任务。别忘了，我们之前调用offer方法存数据的时候，延迟任务就是存储在这里的，score的值我们当时存储的是延迟任务的过期时间。所以,如果score的值小于系统当前时间，说明这个延迟任务已经过期了，可以让消费者取出来了。limit 0 100的意思是，一次最多只取100条数据。

if #expiredValues > 0 then这行代码的意思是，如果能取出来数据，就使用for循环处理这些已经过期的数据。

在for循环里面执行了俩条Redis命令。第一条为:rpush KEYS[1], value翻译过来为:rpush getName() value => rpush orderQueue 用户数据,意思就是将取出来的数据继续放进Redis的List数据结构里面,value就是用户当时存进来的具体数据。这个List里面存放的全是已经过期的数据,take方法就是从这里获取数据的。过期的数据就是在这里放进去的，这个是最核心的代码了。 第二条Redis命令为:lrem KEYS[3] 1 v => lrem queueName 1 v 从queueName里面删除一个数据。因为之前offer方法往这里面存数据了，所以过期之后这里要再把它删除掉。

循环结束之后，又执行了一条命令:zrem KEYS[2] unpack(expiredValues) => zrem timeoutSetName 所有取出来的数据，意思就是把所有的过期数据，从SortedSet这个数据结构中删除掉。

最后再执行zrange KEYS[2] 0 0 WITHSCORES=>zrange timeoutSetName 0 0 withscores意思就是从Redis的SortedSet数据结构取出第一条也就是最先过期的延迟任务。

if v[1] ~= nil then return v[2]; 这行代码的意思是，如果上面的zrange能取出来数据，就把取出来的数据的score值(任务的过期时间)返回出去。

在RedissonDelayedQueue的构造方法会触发任务的调度，在这个任务里面会动态的触发定时任务的执行，这些定时任务会在任务过期时调用pushTaskAsync()方法，执行上面的Redis命令，将过期数据放入目标延迟队列供消费者消费。

调度任务

调度任务开始执行

执行逻辑

注意上图中RTopic schedulerTopic = getTopic();这行代码，调用的就是RedissonDelayedQueue构造方法里面定义的getTopic()方法。

getTopic()方法

pushTask()方法

scheduleTask()方法

总结如下：

在RedissonDelayedQueue类的构造方法里面会直接调用QueueTransferTask这个类的start()方法。

在QueueTransferTask这个类的start()方法里面会给RTopic schedulerTopic = getTopic()其实就是RedissonTopic这个类注册俩个Listener监听事件：1.onSubscribe(订阅监听)2.onMessage(消息监听)

schedulerTopic.addListener()这个方法在添加订阅监听的同时也会主动去订阅一下，订阅完直接就触发onSubscribe(订阅监听)监听事件了。触发onSubscribe(订阅监听)事件之后，onSubscribe方法会调用pushTask()方法，pushTask()方法会调用pushTaskAsync()去执行Redis命令，过期数据就会放到目标队列里面了。

如果pushTaskAsync()方法有返回值(任务过期时间)，会调用scheduleTask(res)方法，scheduleTask(res)方法会启动一个定时任务，定时任务到时间会执行pushTask()方法。

最后一步，验证一下到底是不是schedulerTopic.addListener()这个方法会不会主动触发监听Subscribe这个命令呢？

new RedissonTopic

addListenert添加监听事件

接下来我们，连接上Redis,使用呢monitor命令，监控命令的执行。monitor命令可以监控redis执行了哪些命令，生产环境禁止使用这个命令。

连接Redis执行命令监听

然后创建一个Controller方法，启动项目并访问接口，主要是为了触发一下我们代码的执行，Controller的代码如下

访问这个Controller接口，http://localhost:8080/order/testRedissonDelayQueueTake。 然后去Redis服务器上面看是否监控到SUBSCRIBE订阅命令。

SUBSCRIBE命令

可以看到，访问接口之后Redisson向Redis发出了SUBSCRIBE订阅的命令。所以，其实真正触发SUBSCRIBE订阅命令的是这行代码testTopic.addListener()。

给大家画个流程图吧：

构造方法触发定时任务流程图

offer方法触发定时任务流程图

不知道大家能不能看懂，反正我是已经非常清楚了，我自己看源码看了俩天。我的文章只是提供一些线索，你们根据这一点点小提示、小线索，自己去看源码，很快就明白了。