1.延迟任务的问题提出
上一篇写了使用RabbitMQ来实现延迟任务的实现,其实实现延迟任务的方式有很多,各有利弊,有单机和分布式的。在这里做一个总结,在遇到这类问题的时候希望给大家一个参考和思路。 延迟任务有别于定时任务,定时任务往往是固定周期的,有明确的触发时间。而延迟任务一般没有固定的开始时间,它常常是由一个事件触发的,而在这个事件触发之后的一段时间内触发另一个事件。
延迟任务相关的业务场景如下:
场景一:物联网系统经常会遇到向终端下发命令,如果命令一段时间没有应答,就需要设置成超时。 场景二:订单下单之后30分钟后,如果用户没有付钱,则系统自动取消订单。
下面我们来探讨一些方案,其实这些方案没有好坏之分,和系统架构一样,只有最适合。对于数据量较小的情况下,任意一种方案都可行,考虑的是简单明了和开发速度,尽量避免把系统搞复杂了。而对于数据量较大的情况下,就需要有一些选择,并不是所有的方案都适合了。
2.常用的延迟任务实现方式
2.1 数据库轮询
这是比较常见的一种方式,所有的订单或者所有的命令一般都会存储在数据库中。我们会起一个线程去扫数据库或者一个数据库定时Job,找到那些超时的数据,直接更新状态,或者拿出来执行一些操作。这种方式很简单,不会引入其他的技术,开发周期短。
如果数据量比较大,千万级甚至更多,插入频率很高的话,上面的方式在性能上会出现一些问题,查找和更新对会占用很多时间,轮询频率高的话甚至会影响数据入库。一种可以尝试的方式就是使用类似XXL-JOB或Elastic-Job这样的分布式的任务调度加上数据分片功能,把需要判断的数据分到不同的机器上执行。
如果数据量进一步增大,那扫数据库肯定就不行了。另一方面,对于订单这类数据,我们也许会遇到分库分表,那上述方案就会变得过于复杂,得不偿失。
2.2JDK DelayQueue延迟队列
上一篇文章已经针对JDK DelayQueue延迟队列坐过介绍,这里就不继续阐述了。
2.3 JDK ScheduledExecutorService
JDK自带的一种线程池,它能调度一些命令在一段时间之后执行,或者周期性的执行。文章开头的一些业务场景主要使用第一种方式,即,在一段时间之后执行某个操作。代码例子如下:
package com.yangyang.corejava.task;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* @author chenshunyang
* @create 2018-05-09 17:28
**/
public class ScheduledExecutorServiceTest {
private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:MM:ss");
public static void main(String[] args) {
ScheduledExecutorService executor = Executors.newScheduledThreadPool(100);
for (int i = 10; i > 0; i--) {
executor.schedule(new Runnable() {
public void run() {
System.out.println("Work start, thread id:" + Thread.currentThread().getId() + " " + sdf.format(new Date()));
}
}, i, TimeUnit.SECONDS);
}
}
}
复制
执行结果:
ScheduledExecutorService的实现类ScheduledThreadPoolExecutor提供了一种并行处理的模型,简化了线程的调度。DelayedWorkQueue是类似DelayQueue的实现,也是基于最小堆的、线程安全的数据结构,所以会有上例排序后输出的结果。
ScheduledExecutorService比上面一种DelayQueue更加实用。因为,一般来说,使用DelayQueue获取消息后触发事件都会实用多线程的方式执行,以保证其他事件能准时进行。而ScheduledThreadPoolExecutor就是对这个过程进行了封装,让大家更加方便的使用。同时在加强了部分功能,比如定时触发命令。
2.4 时间轮
时间轮是一种非常惊艳的数据结构。其在Linux内核中使用广泛,是Linux内核定时器的实现方法和基础之一。按使用场景,大致可以分为两种时间轮:原始时间轮和分层时间轮。分层时间轮是原始时间轮的升级版本,来应对时间“槽”数量比较大的情况,对内存和精度都有很高要求的情况。我们延迟任务的场景一般只需要用到原始时间轮就可以了。
原始时间轮:如下图一个轮子,有8个“槽”,可以代表未来的一个时间。如果以秒为单位,中间的指针每隔一秒钟转动到新的“槽”上面,就好像手表一样。如果当前指针指在1上面,我有一个任务需要4秒以后执行,那么这个执行的线程回调或者消息将会被放在5上。那如果需要在20秒之后执行怎么办,由于这个环形结构槽数只到8,如果要20秒,指针需要多转2圈。位置是在2圈之后的5上面(20 % 8 + 1)。这个圈数需要记录在槽中的数据结构里面。这个数据结构最重要的是两个指针,一个是触发任务的函数指针,另外一个是触发的总第几圈数。时间轮可以用简单的数组或者是环形链表来实现。
相比DelayQueue的数据结构,时间轮在算法复杂度上有一定优势。DelayQueue由于涉及到排序,需要调堆,插入和移除的复杂度是O(lgn),而时间轮在插入和移除的复杂度都是O(1)。
时间轮比较好的开源实现是Netty的
// 创建Timer, 精度为100毫秒,
HashedWheelTimer timer = new HashedWheelTimer();
System.out.println(sdf.format(new Date()));
MyTask task1 = new MyTask();
MyTask task2 = new MyTask();
MyTask task3 = new MyTask();
timer.newTimeout(task1, 5, TimeUnit.SECONDS);
timer.newTimeout(task2, 10, TimeUnit.SECONDS);
timer.newTimeout(task3, 15, TimeUnit.SECONDS);
// 阻塞main线程
try {
System.in.read();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
复制
其中HashedWheelTimer有多个构造函数。其中: ThreadFactory :创建线程的类,默认Executors.defaultThreadFactory()。 TickDuration:多少时间指针顺时针转一格,单位由下面一个参数提供。 TimeUnit:上一个参数的时间单位。 TicksPerWheel:时间轮上的格子数。 如果一个任务要在120s后执行,时间轮是默认参数的话,那么这个任务在时间轮上需要经过 120000ms / (512 * 100ms) = 2轮 120000ms % (512 * 100ms) = 176格。 在使用HashedWheelTimer的过程中,延迟任务的实现最好使用异步的,HashedWheelTimer的任务管理和执行都在一个线程里面。如果任务比较耗时,那么指针就会延迟,导致整个任务就会延迟。
2.5Quartz
quartz是一个企业级的开源的任务调度框架,quartz内部使用TreeSet来保存Trigger,如下图。Java中的TreeSet是使用TreeMap实现,TreeMap是一个红黑树实现。红黑树的插入和删除复杂度都是logN。和最小堆相比各有千秋。最小堆插入比红黑树快,删除顶层节点比红黑树慢。
相比上述的三种轻量级的实现功能丰富很多。有专门的任务调度线程,和任务执行线程池。quartz功能强大,主要是用来执行周期性的任务,当然也可以用来实现延迟任务。但是如果只是实现一个简单的基于内存的延时任务的话,quartz就稍显庞大。
2.6. RabbitMQ TTL和DXL
使用RabbitMQ的TTL和DXL实现延迟队列在这里不做详细的介绍,前面的文章已经写到过。
综上所述,解决延迟队列有很多种方法。选择哪个解决方案也需要根据不同的数据量、实时性要求、已有架构和组件等因素进行判断和取舍。对于比较简单的系统,可以使用数据库轮训的方式。数据量稍大,实时性稍高一点的系统可以使用JDK延迟队列(也许需要解决程序挂了,内存中未处理任务丢失的情况)。如果需要分布式横向扩展的话推荐使用quartz的方案。但是对于系统中已有RabbitMQ,那RabbitMQ会是一个更好的方案。
本文参考文章:http://www.cnblogs.com/haoxinyue/p/6663720.html
欢迎关注我的公众号,获取更多文章,并与我交流沟通。