总结：延时任务队列的原理与实现

  • Z时代
  • 2024-01-10
  • 分类：综合

编程

 

延迟任务相关的业务场景如下：

  • 场景一：在订单系统中，一个用户某个时刻下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将自动进行过期处理。

  • 场景二：用户某个时刻通过手机远程遥控家里的智能设备在指定的时间进行工作。这时就可以将用户指令发送到延时队列，当指令设定的时间到了再将指令推送到只能设备。

下面我们来探讨一些方案，其实这些方案没有好坏之分，和系统架构一样，只有最适合。对于数据量较小的情况下，任意一种方案都可行，考虑的是简单明了和开发速度，尽量避免把系统搞复杂了。而对于数据量较大的情况下，就需要有一些选择，并不是所有的方案都适合了。

解决方式

  • 定时器轮询遍历数据库记录

  • JDK的DelayQueue

  • JDK ScheduledExecutorService

  • 时间轮（netty）

  • 利用quartz等定时任务

  • Redis的ZSet实现

  • rabbitMq实现延时队列

定时器轮询遍历数据库记录

这是比较常见的一种方式，所有的订单或者所有的命令一般都会存储在数据库中。我们会起一个线程定时去扫数据库或者一个数据库定时Job，找到那些超时的数据，直接更新状态，或者拿出来执行一些操作。这种方式很简单，不会引入其他的技术，开发周期短。

如果数据量比较大，千万级甚至更多，插入频率很高的话，上面的方式在性能上会出现一些问题，查找和更新对会占用很多时间，轮询频率高的话甚至会影响数据入库。一种可以尝试的方式就是使用类似TBSchedule或Elastic-Job这样的分布式的任务调度加上数据分片功能，把需要判断的数据分到不同的机器上执行。

如果数据量进一步增大，那扫数据库肯定就不行了。另一方面，对于订单这类数据，我们也许会遇到分库分表，那上述方案就会变得过于复杂，得不偿失。

JDK的DelayQueue

Java中的DelayQueue位于java.util.concurrent包下，作为单机实现，它很好的实现了延迟一段时间后触发事件的需求。由于是线程安全的它可以有多个消费者和多个生产者，从而在某些情况下可以提升性能。DelayQueue本质是封装了一个PriorityQueue，使之线程安全，加上Delay功能，也就是说，消费者线程只能在队列中的消息“过期”之后才能返回数据获取到消息，不然只能获取到null。

之所以要用到PriorityQueue，主要是需要排序。也许后插入的消息需要比队列中的其他消息提前触发，那么这个后插入的消息就需要最先被消费者获取，这就需要排序功能。PriorityQueue内部使用最小堆来实现排序队列。队首的，最先被消费者拿到的就是最小的那个。使用最小堆让队列在数据量较大的时候比较有优势。使用最小堆来实现优先级队列主要是因为最小堆在插入和获取时，时间复杂度相对都比较好，都是O(logN)。

下面例子实现了未来某个时间要触发的消息。我把这些消息放在DelayQueue中，当消息的触发时间到，消费者就能拿到消息，并且消费，实现处理方法。示例代码：

/*
 * 定义放在延迟队列中的对象，需要实现Delayed接口
 */
publicclassDelayedTaskimplementsDelayed{
privateint _expireInSecond =0;
publicDelayedTask(int delaySecond){
Calendar cal =Calendar.getInstance();
        cal.add(Calendar.SECOND, delaySecond);
        _expireInSecond =(int)(cal.getTimeInMillis()/1000);
}
publicintcompareTo(Delayed o){
long d =(getDelay(TimeUnit.NANOSECONDS)- o.getDelay(TimeUnit.NANOSECONDS));
return(d ==0)?0:((d <0)?-1:1);
}
publiclonggetDelay(TimeUnit unit){
// TODO Auto-generated method stub
Calendar cal =Calendar.getInstance();
return _expireInSecond -(cal.getTimeInMillis()/1000);
}
}

下面定义了三个延迟任务，分别是10秒，5秒和15秒。依次入队列，期望5秒钟后，5秒的消息先被获取到，然后每个5秒钟，依次获取到10秒数据和15秒的那个数据。

publicstaticvoidmain(String[] args) throws InterruptedException {
// TODO Auto-generated method stub
SimpleDateFormat sdf =newSimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//定义延迟队列
        DelayQueue<DelayedTask> delayQueue =newDelayQueue<DelayedTask>();
//定义三个延迟任务
DelayedTask task1 =newDelayedTask(10);
DelayedTask task2 =newDelayedTask(5);
DelayedTask task3 =newDelayedTask(15);
        delayQueue.add(task1);
        delayQueue.add(task2);
        delayQueue.add(task3);
        System.out.println(sdf.format(newDate())+" start");
while(delayQueue.size()!=0){
//如果没到时间，该方法会返回
DelayedTask task = delayQueue.poll();
if(task !=null){
Date now =newDate();
                System.out.println(sdf.format(now));
}
            Thread.sleep(1000);
}
}

DelayQueue是一种很好的实现方式，虽然是单机，但是可以多线程生产和消费，提高效率。拿到消息后也可以使用异步线程去执行下一步的任务。如果有分布式的需求可以使用Redis来实现消息的分发，如果对消息的可靠性有非常高的要求可以使用消息中间件.

JDK ScheduledExecutorService

JDK自带的一种线程池，它能调度一些命令在一段时间之后执行，或者周期性的执行。文章开头的一些业务场景主要使用第一种方式，即，在一段时间之后执行某个操作。代码例子如下：

publicstaticvoidmain(String[] args){
// TODO Auto-generated method stub
ScheduledExecutorService executor = Executors.newScheduledThreadPool(100);
for(int i =10; i >0; i--){
            executor.schedule(newRunnable(){
publicvoidrun(){
// TODO Auto-generated method stub
                    System.out.println(
"Work start, thread id:"+ Thread.currentThread().getId()+" "+ sdf.format(newDate()));
}
}, i, TimeUnit.SECONDS);
}
}

ScheduledExecutorService的实现类ScheduledThreadPoolExecutor提供了一种并行处理的模型，简化了线程的调度。DelayedWorkQueue是类似DelayQueue的实现，也是基于最小堆的、线程安全的数据结构，所以会有上例排序后输出的结果。

ScheduledExecutorService比上面一种DelayQueue更加实用。因为，一般来说，使用DelayQueue获取消息后触发事件都会实用多线程的方式执行，以保证其他事件能准时进行。而ScheduledThreadPoolExecutor就是对这个过程进行了封装，让大家更加方便的使用。同时在加强了部分功能，比如定时触发命令。

时间轮

时间轮是一种非常惊艳的数据结构。其在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 =newHashedWheelTimer();
        System.out.println(sdf.format(newDate()));
MyTask task1 =newMyTask();
MyTask task2 =newMyTask();
MyTask task3 =newMyTask();
        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的任务管理和执行都在一个线程里面，任务会耗时，那么指针就会延迟，导致整个任务就会延迟。

Quartz

quartz是一个企业级的开源的任务调度框架，quartz内部使用TreeSet来保存Trigger，如下图。Java中的TreeSet是使用TreeMap实现，TreeMap是一个红黑树实现。红黑树的插入和删除复杂度都是logN。和最小堆相比各有千秋。最小堆插入比红黑树快，删除顶层节点比红黑树慢。
相比上述的三种轻量级的实现功能丰富很多。有专门的任务调度线程，和任务执行线程池。quartz功能强大，主要是用来执行周期性的任务，当然也可以用来实现延迟任务。但是如果只是实现一个简单的基于内存的延时任务的话，quartz就稍显庞大。

Redis ZSet

Redis中的ZSet是一个有序的Set，内部使用HashMap和跳表(SkipList)来保证数据的存储和有序，HashMap里放的是成员到score的映射，而跳跃表里存放的是所有的成员，排序依据是HashMap里存的score,使用跳跃表的结构可以获得比较高的查找效率，并且在实现上比较简单。

publicclassZSetTest{
privateJedisPool jedisPool =null;
// Redis服务器IP
privateString ADDR ="10.23.22.42";
// Redis的端口号
privateint PORT =6379;
privateSimpleDateFormat sdf =newSimpleDateFormat("yyyy-MM-dd HH:mm:ss");
publicvoidintJedis(){
        jedisPool =newJedisPool(ADDR, PORT);
}
publicstaticvoidmain(String[] args){
// TODO Auto-generated method stub
ZSetTest zsetTest =newZSetTest();
        zsetTest.intJedis();
        zsetTest.addItem();
        zsetTest.getItem();
        zsetTest.deleteZSet();
}
publicvoiddeleteZSet(){
Jedis jedis = jedisPool.getResource();
        jedis.del("zset_test");
}
publicvoidaddItem(){
Jedis jedis = jedisPool.getResource();
Calendar cal1 = Calendar.getInstance();
        cal1.add(Calendar.SECOND,10);
int second10later =(int)(cal1.getTimeInMillis()/1000);
Calendar cal2 = Calendar.getInstance();
        cal2.add(Calendar.SECOND,20);
int second20later =(int)(cal2.getTimeInMillis()/1000);
Calendar cal3 = Calendar.getInstance();
        cal3.add(Calendar.SECOND,30);
int second30later =(int)(cal3.getTimeInMillis()/1000);
Calendar cal4 = Calendar.getInstance();
        cal4.add(Calendar.SECOND,40);
int second40later =(int)(cal4.getTimeInMillis()/1000);
Calendar cal5 = Calendar.getInstance();
        cal5.add(Calendar.SECOND,50);
int second50later =(int)(cal5.getTimeInMillis()/1000);
        jedis.zadd("zset_test", second50later,"e");
        jedis.zadd("zset_test", second10later,"a");
        jedis.zadd("zset_test", second30later,"c");
        jedis.zadd("zset_test", second20later,"b");
        jedis.zadd("zset_test", second40later,"d");
        System.out.println(sdf.format(newDate())+" add finished.");
}
publicvoidgetItem(){
Jedis jedis = jedisPool.getResource();
while(true){
try{
                Set<Tuple>set= jedis.zrangeWithScores("zset_test",0,0);
Stringvalue=((Tuple)set.toArray()[0]).getElement();
int score =(int)((Tuple)set.toArray()[0]).getScore();
Calendar cal = Calendar.getInstance();
int nowSecond =(int)(cal.getTimeInMillis()/1000);
if(nowSecond >= score){
                    jedis.zrem("zset_test",value);
                    System.out.println(sdf.format(newDate())+" removed value:"+value);
}
if(jedis.zcard("zset_test")<=0)
{
                    System.out.println(sdf.format(newDate())+" zset empty ");
return;
}
                Thread.sleep(1000);
}catch(InterruptedException e){
// TODO Auto-generated catch block
                e.printStackTrace();
}
}
}
}

在用作延迟任务的时候，可以在添加数据的时候，使用zadd把score写成未来某个时刻的unix时间戳。消费者使用zrangeWithScores获取优先级最高的（最早开始的的）任务。注意，zrangeWithScores并不是取出来，只是看一下并不删除，类似于Queue的peek方法。程序对最早的这个消息进行验证，是否到达要运行的时间，如果是则执行，然后删除zset中的数据。如果不是，则继续等待。

由于zrangeWithScores 和 zrem是先后使用，所以有可能有并发问题，即两个线程或者两个进程都会拿到一样的一样的数据，然后重复执行，最后又都会删除。如果是单机多线程执行，或者分布式环境下，可以使用Redis事务，也可以使用由Redis实现的分布式锁，或者使用下例中Redis Script。你可以在Redis官方的Transaction章节找到事务的相关内容。

使用Redis的好处主要是：

  • 解耦：把任务、任务发起者、任务执行者的三者分开，逻辑更加清晰，程序强壮性提升，有利于任务发起者和执行者各自迭代，适合多人协作。

  • 异常恢复：由于使用Redis作为消息通道，消息都存储在Redis中。如果发送程序或者任务处理程序挂了，重启之后，还有重新处理数据的可能性。

  • 分布式：如果数据量较大，程序执行时间比较长，我们可以针对任务发起者和任务执行者进行分布式部署。特别注意任务的执行者，也就是Redis的接收方需要考虑分布式锁的问题。

RabbitMQ TTL和DXL

AMQP和RabbitMQ本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。
但是我们可以通过RabbitMQ的两个特性来曲线实现延迟队列：

  • Time To Live(TTL)
    RabbitMQ可以针对Queue和Message设置 x-message-tt，来控制消息的生存时间，如果超时，则消息变为dead letter
    RabbitMQ针对队列中的消息过期时间有两种方法可以设置。

    • A: 通过队列属性设置，队列中所有消息都有相同的过期时间。

    • B: 对消息进行单独设置，每条消息TTL可以不同。
      如果同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就成为dead letter

  • Dead Letter Exchanges（DLX）
    RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由。

    • x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange

    • x-dead-letter-routing-key：指定routing-key发送

    • 队列出现dead letter的情况有：

      • 消息或者队列的TTL过期

      • 队列达到最大长度

      • 消息被消费端拒绝（basic.reject or basic.nack）并且requeue=false

利用DLX，当消息在一个队列中变成死信后，它能被重新publish到另一个Exchange。这时候消息就可以重新被消费。

Ref：
http://www.mamicode.com/info-detail-1745315.html

以上是 总结：延时任务队列的原理与实现 的全部内容， 来源链接： utcz.com/z/512967.html

回到顶部