Redisson分布式延时队列 RedissonDelayedQueue原创
1 先回顾一下Java原生自带的DelayQueue
-
自己实现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的代码示例如下：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class DelayQueueDemo {
static DelayQueue<Delayed> queue = new DelayQueue();
public static void main(String[] args) throws InterruptedException {
queue.add(new MyDelay(100, TimeUnit.SECONDS, "第一次添加任务"));
queue.add(new MyDelay(1, TimeUnit.SECONDS, "第二次添加任务"));
queue.add(new MyDelay(5, TimeUnit.SECONDS, "第三次添加任务"));
queue.add(new MyDelay(10000, TimeUnit.MILLISECONDS, "第四次添加任务，只有到了指定的延迟时间才能调用queue.take()方法，把这个任务取出来"));
while(!queue.isEmpty()){
// queue.take()从延迟队列中取出任务，如果任务指定的延迟时间还没有到，这里是取不出来的，线程将一直阻塞
// 线程状态将处于java.lang.Thread.State: TIMED_WAITING (parking),会释放CPU,底层调用的是 UNSAFE.park方法。
Delayed delayed = queue.take();
System.out.println("这么快就取出来了？？" + delayed);
}
}
}
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class MyDelay<T> implements Delayed {
// 延迟时间,(时间单位会在计算剩余时间的方法getDelay里面,由你自己指定，一般来说都会使用毫秒,更精确一点。)
long delayTime;
// 过期时间,(时间单位会在计算剩余时间的方法getDelay里面,由你自己指定，一般来说都会使用毫秒,更精确一点。)
long expire;
// 你自己放进队列里的数据
T data;
public MyDelay(long delayTime, TimeUnit delayTimeUnit, T t) {
// 将用户传进来的时间转换为毫秒
this.delayTime = TimeUnit.MILLISECONDS.convert(delayTime, delayTimeUnit);
// 过期时间 = 当前时间 + 延迟时间(时间单位会在计算剩余时间的方法getDelay里面,由你自己指定，一般来说都会使用毫秒,更精确一点。)
// 当然你也可以使用别的时间,随意的
this.expire = System.currentTimeMillis() + this.delayTime;
data = t;
}
/**
* 剩余时间 = 过期时间 - 当前时间
*
*/
@Override
public long getDelay(TimeUnit unit) {
// 注意convert这个方法,第一个参数是一个long类型的数值,第二个参数的意思是告诉convert第一个long类型的值的单位是毫秒
return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
/**
* 优先级:俩个任务比较，时间短的优先执行
*
*/
@Override
public int compareTo(Delayed o){
long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
return (int)f;
}
@Override
public String toString() {
// 这个toString()方法不是必须的，你可以不重写。写不写都无所谓，我这里为了测试，将数据打印出来了。
return "delayTime=" + delayTime + ",expire=" + expire + ",data=" + data;
}
}
- 源码分析参考这篇文章《DelayQueue延迟队列》
Redisson延迟队列
少废话，先看代码
- pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.redisson</groupId>
<artifactId>demo</artifactId>
<version>1.0-SNAPSHOT</version>
<description>Demo project for Spring Boot</description>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.1.RELEASE</version>
<!-- lookup parent from repository -->
<relativePath/>
</parent>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.10.7</version>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
</dependencies>
</project>
- SpringBoot的application.yml配置文件
server:
port: 8080
spring:
redis:
host: 192.168.212.132
port: 6379
password: 123321
application:
name: redissonTest
- Redisson配置类:RedissonConfig.java
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SentinelServersConfig;
import org.redisson.config.SingleServerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RedissonConfig {
@Value("${spring.application.name}")
private String serverName;
@Bean
public RedissonClient redisson(RedisProperties redisProperties) {
Config config = new Config();
SingleServerConfig singleServerConfig = config.useSingleServer();
singleServerConfig.setAddress("redis://" + redisProperties.getHost() + ":" + redisProperties.getPort());
singleServerConfig.setPassword(redisProperties.getPassword());
singleServerConfig.setKeepAlive(true);
singleServerConfig.setDatabase(redisProperties.getDatabase());
singleServerConfig.setClientName(serverName);
return Redisson.create(config);
}
}
- 使用Redisson添加任务到延迟队列里面
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
@Component
public class AddTaskToDelayQueue {
@Autowired
RedissonClient redissonClient;
/**
* 添加任务到延时队列里面
*
* @param orderId 订单ID
*/
public void addTaskToDelayQueue(String orderId) {
// RBlockingDeque的实现类为:new RedissonBlockingDeque
RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque("orderQueue");
// RDelayedQueue的实现类为:new RedissonDelayedQueue
RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "添加任务到延时队列里面");
delayedQueue.offer(orderId +"添加一个任务", 3, TimeUnit.SECONDS);
delayedQueue.offer(orderId +"添加二个任务", 6, TimeUnit.SECONDS);
delayedQueue.offer(orderId +"添加三个任务", 9, TimeUnit.SECONDS);
}
}
- Controller代码
import com.redisson.demo.AddTaskToDelayQueue;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
@RestController
@RequestMapping("order")
public class TestController {
@Autowired
private RedissonClient redissonClient;
@Autowired
AddTaskToDelayQueue addTaskToDelayQueue;
@GetMapping("testRedissonDelayQueueTake")
public void testRedissonDelayQueueTake() {
RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque("orderQueue");
// 注意虽然delayedQueue在这个方法里面没有用到，但是这行代码也是必不可少的。
RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
while (true) {
String orderId = null;
try {
orderId = blockingDeque.take();
} catch (Exception e) {
System.err.println(e.getStackTrace());
continue;
}
if (orderId == null) {
continue;
}
System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "延时队列收到:" + orderId);
}
}
@GetMapping("testRedissonDelayQueueOffer")
public void testRedissonDelayQueueOffer() {
addTaskToDelayQueue.addTaskToDelayQueue("testRedissonDelayQueue");
}
}
解析代码
我们先看添加数据delayedQueue.offer()方法的源码
- **offer()**方法是往Redisson的RDelayedQueue延迟队列里面添加延迟任务的，所以先看这个方法。
// RBlockingDeque的实现类为:new RedissonBlockingDeque
RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque("orderQueue");
// RDelayedQueue的实现类为:new RedissonDelayedQueue
RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "添加任务到延时队列里面");
delayedQueue.offer(orderId +"999999999999999111111111111111", 90, TimeUnit.SECONDS);
delayedQueue.offer(orderId +"999999999999992222222222", 180, TimeUnit.SECONDS);
delayedQueue.offer(orderId + "9999999999999333333333333", 280, TimeUnit.SECONDS);
在IDEA里面按住鼠标左键点击redissonClient.getDelayedQueue(blockingDeque)方法，很容易知道RDelayedQueue的实现类为:org.redisson.RedissonDelayedQueue.java。那么offer方法的源码也在org.redisson.RedissonDelayedQueue.java里面，如下截图：
其实，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方法流程是一样的。
接下来看取数据blockingDeque.take()方法的源码
- **take()**方法是从Redisson的RDelayedQueue延迟队列里面取出延迟任务的。
RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque("orderQueue");
// 注意虽然delayedQueue在这个方法里面没有用到，但是这行代码也是必不可少的。
RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
String orderId = blockingDeque.take();
take()这个方法的核心原理非常简单，核心代码就是使用Redis的BLPOP命令，从Redis的List数据结构里面取数据，取不到就阻塞，一直等到有数据进来。源码位置:org.redisson.RedissonBlockingDeque，源码如下截图：
所以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的左边开始消费的。一唱一和，非常完美。
那过期的数据是怎么放进Redis的List数据结构里面的呢？
- 还记得下面这段代码吗？
RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque("orderQueue");
// 注意虽然delayedQueue在这个方法里面没有用到，但是这行代码也是必不可少的。
RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
注意:上面这俩行代码，无论你是在调用offer方法存放数据还是调用take()方法取出数据，这俩行代码都是必不可少的。
过期的数据就是在这俩行代码里面放进去的，核心代码在这里：org.redisson.Redisson
redissonClient.getDelayedQueue(blockingDeque);
源码如下截图：
redissonClient.getDelayedQueue(blockingDeque)方法调用了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值(任务的过期时间)返回出去。
那过期的数据是在什么时候放进Redis的List数据结构里面的呢？
- 在RedissonDelayedQueue的构造方法会触发任务的调度，在这个任务里面会动态的触发定时任务的执行，这些定时任务会在任务过期时调用pushTaskAsync()方法，执行上面的Redis命令，将过期数据放入目标延迟队列供消费者消费。
注意上图中RTopic schedulerTopic = getTopic();这行代码，调用的就是RedissonDelayedQueue构造方法里面定义的getTopic()方法。
总结如下：
-
在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这个命令呢？
接下来我们，连接上Redis,使用呢monitor命令，监控命令的执行。monitor命令可以监控redis执行了哪些命令，生产环境禁止使用这个命令。
然后创建一个Controller方法，启动项目并访问接口，主要是为了触发一下我们代码的执行，Controller的代码如下
import org.redisson.Redisson;
import org.redisson.RedissonTopic;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.api.listener.BaseStatusListener;
import org.redisson.api.listener.MessageListener;
import org.redisson.client.codec.LongCodec;
import org.redisson.command.CommandExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
@RestController
@RequestMapping("order")
public class TestController {
@Autowired
private RedissonClient redissonClient;
@GetMapping("testRedissonDelayQueueTake")
public void testRedissonDelayQueueTake() {
RedissonTopic testTopic = new RedissonTopic(LongCodec.INSTANCE, ((Redisson)redissonClient).getCommandExecutor(), "delayQueueChannelTest");
int statusListenerId = testTopic.addListener(new BaseStatusListener() {
@Override
public void onSubscribe(String channel) {
// 搞一个空方法就行
// pushTask();
}
});
}
}
访问这个Controller接口，http://localhost:8080/order/testRedissonDelayQueueTake。
然后去Redis服务器上面看是否监控到SUBSCRIBE订阅命令。
可以看到，访问接口之后Redisson向Redis发出了SUBSCRIBE订阅的命令。所以，其实真正触发SUBSCRIBE订阅命令的是这行代码testTopic.addListener()。
给大家画个流程图吧：
不知道大家能不能看懂，反正我是已经非常清楚了，我自己看源码看了俩天。我的文章只是提供一些线索，你们根据这一点点小提示、小线索，自己去看源码，很快就明白了。
