深入探讨延时任务解决方案：时间轮算法、Redis延时队列与Java DelayQueue
目录时间轮算法时间轮Redis延迟队列mq延迟队列具体实现总结
概述
延迟任务相信大家都很熟悉，在实际业务中也有很多应用场景。比如下单后15分钟内未付款就会取消订单、外卖超过时限自动赔付等等。在这些情况下，我们的服务的实现应该如何设计呢？
更简单的方法当然是安排一个任务来轮询数据库。然而，当业务量较大、事件处理耗时较长时，我们的系统和数据库往往会面临巨大的压力。如果采用这种方法，数据库和系统可能会崩溃。那么有什么好的办法吗？今天给大家介绍几种实现延迟任务的方法。
JAVA延迟队列
你没看错，Java有一个内置的延迟队列，位于java并发包中。
DelayQueue是jdk自带的延迟队列实现。它的实现依赖于可重入锁ReentrantLock、条件锁Condition和优先级队列PriorityQueue。本质上它也是一个阻塞队列。那么他是如何实现延时效果的呢？
DelayQueue的实现原理
首先，DelayQueue队列中的元素必须继承一个名为Delayed的接口。我们找到这个类
public interface Delayed extends Comparable{ long getDelay(TimeUnit unit); }
发现这个类定义了一个方法getDelay，返回值为long。该方法用于定义队列中元素的过期时间。所有需要放入队列的元素都必须实现该方法。
那么我们来看看延迟队列是如何运作的。让我们看一下最典型的报价并采取：
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
landroidock.lock();
try {
q.offer(e);
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
报价操作平淡无奇。它甚至直接调用优先级队列的offer来根据延迟对队列进行排序。它只是添加一个锁并进行一些数据调整。没有什么深入的东西，但是 take 的实现看起来很复杂。 （注意Dalayed继承了Comparable方法，所以可以直接使用优先级队列进行排序，只要你自己实现compareTo方法）我尝试添加一些注释以使其更清晰：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 自选操作
for (;;) {
// 获取队列第一个元素，如果队列为空
// 阻塞住直到有新元素加入队列，offer等方法调用signal唤醒线程
E first = q.peek();
if (first == null)
available.await();
else {
// 如果队列中有元素
long delay = first.getDelay(NANOSECONDS);
// 判断延时时间，如果到时间了，直接取出数据并return
if (delay <= 0)
return q.poll();
first = null;
// 如果leader为空则阻塞
if (leader != null)
available.await();
else {
// 获取当前线程
Thread thisThread = Thread.currentThread();
DxGiHpBMYH // 设置leader为当前线程
leader = thisThread;
try {
// 阻塞延时时间
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
我们可以看到take的实现依赖于无限旋转。直到第一个队列元素超过超时时间后才会返回。否则，等待他的人只会被挡住。
DelayQueue实现延迟队列的优缺点
读完源码，我们应该对DelayQueue的实现有一个大概的了解，并且对其优缺点有一定的了解。他的优点很明显：
但他的缺点也很明显：
那么延迟队列有没有更好的实现呢？我们继续往下看吧~
时间轮算法
时间轮算法是一种旨在处理延迟任务的算法。现实生活中的应用可以在kafka、netty等项目中找到类似的实现。
时间轮的具体实现
所谓时间轮，顾名思义，就是一个类似时钟的结构，即它的主要结构是一个环形数组，如图：
循环数组一一存储链表。链表存储需要执行的任务。我们在数组中设置执行间隔。假设我们的循环数组的长度为60，每个数组的执行间隔为1s。然后我们每隔1秒就会执行数组下一个元素的链表中的元素。如果是这样的话，那么我们就无法处理超过60秒的延迟任务，这显然是不合适的，所以我们会给每个任务添加一个参数number ofturns来表示该任务将执行多少轮。如果我们有一个150s后执行的任务，那么它应该在30s的位置，圈数应该是2。每次执行链表中的任务时，我们都会取出需要执行的任务在当前循环中执行，然后从链表中删除。如果当前循环没有执行该任务，则修改其圈数，将圈数减1。这样一个简单的时间轮就创建了。
那么这样的时间轮有哪些优点和缺点呢？
先说优点：
当然，他也有很多缺点：
时间轮算法进阶优化版
刚才提到了时间轮算法的一些缺点，那么有没有一些方法可以对其进行优化呢？这里我就介绍一下时间轮的优化版本。
之前我们提到过，不同轮数的任务会在同一个链表中重复遍历，影响效率。针对这种情况，我们可以进行如下优化： 将时间轮分层
从图中我们可以看到我们采用了多层次的设计。上图分为三层。每层有 60 个单元。第一个轮盘赌的间隔是1小时。我们的数据每次都会插入。在这个轮盘中，每当这个轮盘在一小时后达到下一个刻度时，就会根据延迟时间将其中的所有元素取出并放入象征分钟的第二个轮盘中，以此类推。
这样实现的好处可以说是显而易见的：
时间轮算法的应用
你以前可能没有听说过时间轮算法，但它在各个地方发挥着重要作用。时间轮包含在Linux定时器的实现中。同样，如果你是一个喜欢阅读源码的读者，你也可能会发现它在kafka和netty中的实现。
卡夫卡
kafka中应用了时间轮算法。它的实现与之前提到的高级版时间轮没有太大区别。只有一点：kafka内部实现的时间轮应用到了DelayQueue上。
@nonthreadsafe
private[timer] class TimingWheel(tickMs: Long, wheelSize: Int, startMs: Long, taskCounter: AtomicInteger, queue: DelayQueue[TimerTaskList]) {
private[this] val interval = tickMs * wheelSize
private[this] val buckets = Array.tabulate[TimerTaskList](wheelSize) { _ => new TimerTaskList(taskCounter) }
private[this] var currentTime = startMs - (startMs % tickMs)
@volatile private[this] var overflowWheel: TimingWheel = null
private[this] def addOverflowWheel(): Unit = {
synchronized {
if (overflowWheel == null) {
overflowWheel = new TimingWheel(
tickMs = interval,
wheelSize = wheelSize,
startMs = currentTime,
taskCounter = taskCounter,
queue
)
}
}
}
def add(timerTaskEntry: TimerTaskEntry): Boolean = {
val expiration = timerTaskEntry.expirationMs
if (timerTaskEntry.cancelled) {
false
} else if (expiration < currentTime + tickMs) {
false
} else if (expiration < currentTime + interval) {
val virtualId = expiration / tickMs
val bucket = buckets((virtualId % wheelSize.toLong).toInt)
bucket.add(timerTaskEntry)
if (bucket.setExpiration(virtualId * tickMs)) {
queue.offer(bucket)
}
tjsrue
} else {
if (overflowWheel == null) addOverflowWheel()
overflowWheel.add(timerTaskEntry)
}
}
def advanceClock(timeMs: Long): Unit = {
if (timeMs >= currentTime + tickMs) {
currentTime = timeMs - (timeMs % tickMs)
if (overflowWheel != null) overflowWheel.advanceClock(currentTime)
}
}
}
以上是kafka的内部实现（使用的语言是Scala）。我们可以看到实现非常简单，使用的是DelayQueue。刚才我们讨论了DelayQueue的优点和缺点。查看源码后，我们已经可以得出一个大概的结论：Kafka的时间轮中DelayQueue的作用就是推动任务，以防止任务在时间轮中进行比较。稀疏造成的“空进”。 DelayQueue的触发机制可以很好的避免这种情况。同时，由于DelayQueue的插入效率较低，因此仅用于底层推进。任务的插入是通过时间轮来操作的。两者的配置可以达到效率和资源的平衡。
网蒂
Netty内部还有一个时间轮的实现HashedwheelTimer
HashedWheelTimer的实现比Kafka的内部实现要复杂得多。与Kafka不同的是，它的内部推进并不依赖于DelayQueue而是实现了自己的一套。源代码太长了。有兴趣的读者可以自己看一下。
概括
时间轮说了这么多，我们可以看到它的效率是突出的，但是仍然存在一个问题：它不支持分发。当我们的业务非常复杂，需要分布式的时候，时间轮就显得不够用了。那么此时有什么更好的延迟队列选项呢？我们可能会尝试使用第三方工具
redis延迟队列
其实说到延迟，如果我们经常使用redis的话，我们就会记得redis有一个过期机制，那么我们是否可以利用这个机制来实现一个延迟队列呢？
Redis有自己的key过期机制，过期后可以设置回调方法。基于这个特性，我们可以轻松完成一个延迟队列。当任务进来时，设置定时时间并配置过期回调方法。
除了利用redis的过期机制之外，我们还可以利用其自带的zset来实现延迟队列。 zset支持高性能排序，所以当我们的任务进来时，我们可以使用时间戳作为排序的依据，这样就可以有序地安排任务的执行，这样也可以实现延迟队列。
zset实现延迟队列的好处：
mq延迟队列rocketmq延迟消息
RocketMQ天然支持延迟消息。其延迟消息分为18级，每级对应不同的延迟时间。
那么他的原则是什么？
RocketMQ的broker收到消息后，会将消息写入commitlog，并判断该消息是否为延迟消息（即delay属性是否大于0）。如果确定确实是延迟消息，则不会立即写入，而是通过转发将消息放入对应的延迟主题中（18个延迟级别对应18个主题）
RocketMQ会有一个定时任务进行轮询。如果任务的延迟时间已过，则会将其发送到指定的主题。
这种设计比较简单粗暴，但其缺点也非常明显：
RocketMQ的精确延迟消息
RocketMQ本身不支持精确延迟，但其商业版本ONS支持。不过rocketmq社区里也有相应的解决方案。该解决方案是借助时间轮算法来实现的。有兴趣的朋友可以去社区看看。 （社区中一些未合并的PR是很好的实现参考）
总结
延迟队列的实现有数千种，但如果要在生产中大规模使用，大多数情况下都无法避免时间轮算法。改进后的时间轮算法可以做到精准延迟、持久化、高性能、高可用，堪称完美。但话又说回来，其他的延迟方法就没用了吗？其实不是，所有的方法都需要匹配自己的使用场景。如果您轮询的数据量非常小，那么定期轮询数据库可能是最好的解决方案，而不是盲目地引入复杂的延迟队列。如果是单机任务，那么jdk的延迟队列也是一个不错的选择。
本文介绍的延迟队列只是为了向大家展示它们的原理、优缺点，具体使用需要结合自己的业务场景。
phper