沈阳市住房和城乡建设局网站首页互联网站

大家好，我是一名CRUD工程师，最近我朋友突然来问我如何实现延时队列，我脱口而出就是MQ。不过突然想到公司的项目好像用的是java的一个原生类。于是我就想着趁周末的时间好好的去探究一下各方法实现延时队列的优缺点。

延迟消息

延迟消息就是字面上的意思就是当系统接收到消息之后，需要隔一段时间进行处理，不管是几秒，几分钟还是几个小时，在这的消息发生就叫延时消息

在我不断的进行探究下发现一共有5种常见的方法去实现（欢迎补充哈）

DelayQueue

作为Java的原生类DelayQueue供我们去实现延迟发送。
DelayQueue是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。这种队列是有序的，即队头对象的延迟到期时间最长。

在使用的时候，我们add进去的队列的元素需要实现Delayed接口（同时该接口继承了Comparable接口，所以我们DelayQueue是有序的）
不过这样就会有一个问题，因为DelayQueue它本身是java里的类吗，它是没有持久化的，一旦服务器重启就会导致数据丢失。并且如果进行了多机部署还需要添加分布式锁，所以在我看来在生产上用这个方法不是一个很好的选择，如果延时发送的消息重要性并不是很高那影响就不大。当然作为java的原生类也是有优点的，就是系统不需要和其他服务进行数据通讯，所有的请求都在项目内容进行，就避免了两个服务之间因为信道的不稳定导致的数据丢失的情况。

时间轮算法

具体的介绍可以自行百度
Netty 包里提供了一种时间轮的实现——HashedWheelTimer，其底层使用了数组+链表的数据结构：

//1996 年 George Varghese 和 Tony Lauck 的论文《Hashed and Hierarchical Timing Wheels: 
//Data Structures for the Efficient Implementation of a Timer Facility》中提出了一种时间轮管理 Timeout 事件的方式。其设计非常巧妙，并且类似时钟的运行，
//原始时间轮有 8 个格子，假定指针经过每个格子花费时间是 1 个时间单位，当前指针指向 0，一个 17 个时间单位后超时的任务则需要运转 2 圈再通过一个格子后被执行，放在相同格子的任务会形成一个链表。
public class Test{public static void main(String[] args) {//设置每个格子是 100ms, 总共 256 个格子HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(100, TimeUnit.MILLISECONDS, 256);//加入三个任务，依次设置超时时间是 10s 5s 20sSystem.out.println("加入第一个任务, time= " + LocalDateTime.now());hashedWheelTimer.newTimeout(timeout -> {System.out.println("执行第一个任务, time= " + LocalDateTime.now());}, 10, TimeUnit.SECONDS);System.out.println("加入第二个任务, time= " + LocalDateTime.now());hashedWheelTimer.newTimeout(timeout -> {System.out.println("执行第二个任务, time= " + LocalDateTime.now());}, 5, TimeUnit.SECONDS);System.out.println("加入第三个任务, time= " + LocalDateTime.now());hashedWheelTimer.newTimeout(timeout -> {System.out.println("执行第三个任务, time= " + LocalDateTime.now());}, 20, TimeUnit.SECONDS);System.out.println("等待任务执行===========");}
}

不过这一样会导致数据的丢失，可以在一些不那么重要的情况下使用

Redis

具有存储功能，能够快速读写并具有持久化操作的我一下子就想到了Redis

关于Redis去实现延时队列，我想到了两种方法：

1、Redis里提供了一种数据结构叫做zset，它是可排序的集合并且Redis支持数据持久化。有赞的延迟队列就是基于通过zset进行设计和存储的。
概述：将时间作为zset的分值，zrangewithscores可以获取zset种score值最小的元素（也就是即将到期的任务,去判断系统时间和score的大小，如果相等就执行并删除该任务。（如果想要异步可以使用Timer开一个线程去监听Redis的zset）
2、使用Redis存储数据的过期时间，服务端开启一个过期回调。（较为简单，但在Redis的过期回调中无法获取Key值就需要再Value中再存放一个Key）

消息队列（RabbitMQ的延时队列）

RabbitMQ这个大名我相信大家都听过，在我印象里RabbitMQ自己本身好像是不支持延时发送的，想要实现这个功能主要还是依靠它的TTL（Time To Live 消息存活的时间）
简述：生产者通过Key将消息投入到对应的队列中，但并不对该队列进行消费，等队列里的元素触发了过期时（过期时间就是需要延迟的时间）该消息就会进入死信队列中，此时我们可以将该消息再次转发到正常的队列中进行消费，或者直接在该死信队列中进行消费，从而达到延迟队列的效果。

由于RabbitMQ是专门做消息队列所以它对消息的可靠性会比Redis更加高（消息投递的可靠性、至少处理一次的消费语义，重复投递，手动ACK，投递失败的消息回调等）

消息队列（RocketMQ延时等级）

RocketMQ还可以通过投递消息的时候设置延迟等级

Message message = new Message("test", ("Hello world").getBytes());
//设置延时等级
message.setDelayTimeLevel(3);
producer.send(message);

默认支持18个延迟等级

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

当我们设置了延迟等级的消息之后，RocketMQ不会把消息直接投递到对应的topic，而是转发到对应延迟等级的队列中。在Broker内部会为每个延迟队列起TimerTask来进行判断是否有消息到达了时间。如果到期了，则将消息重新存储到CommitLog，转发到真正目标的topic

结论

这次主要是介绍了java中延时队列各种方法的视线方法，就放了一些的代码（详细的代码会在后面几周发出来）
在公司的项目中其实很多时候并不是越复杂越牛逼的技术越好，我们需要根据不同的业务场景，资源情况去做一个选择。

引用掘金的一句话：
很多时候，我们看到的系统很烂，技术栈很烂，发现好多场景都没有用到最佳实践而感到懊恼，在年轻的时候都想有重构的心。但实际上每引入一个中间件都是需要付出成本的，粗糙也有粗糙的好处。
只要业务能完美支持，那就是好的方案。