每日一博 - 延时任务的多种实现方式解读

古城微笑少年丶 2022-09-08 14:54 190阅读 0赞

### 文章目录 ###

*  Pre
 *  延时任务 VS 定时任务
 *  Solutions
 *   *  DB 轮询
     *   *  核心思想
         *  Demo Code
         *  优缺点
     *  JDK的Delay Queue
     *   *  核心思想
         *  Demo Code
         *  优缺点
     *  时间轮算法
     *   *  核心思想
         *  Demo Code
         *  优缺点
     *  Redis缓存（zset）
     *   *  核心思想
         *  Demo Code
     *  Redis缓存（Keyspace Notifications）
     *   *  核心思想
     *  Redisson 延时队列
     *  使用消息队列

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70]

--------------------

# Pre #

[每日一博 - 使用环形队列实现高效的延时消息][-]

--------------------

# 延时任务 VS 定时任务 #

![在这里插入图片描述][7b841a624f6649d0a3d1330ef0527c47.png]

举个例子，开发中常见的延时任务场景：

*  半小时未支付，取消订单
 *  …

延时任务和定时任务的几个小区别，梳理下：

*  定时任务有明确的触发时间，延时任务没有
 *  定时任务有执行周期，而延时任务在某事件触发后一段时间内执行，没有执行周期
 *  定时任务一般执行的是批处理操作是多个任务，而延时任务一般是单个任务
 *  …

--------------------

# Solutions #

## DB 轮询 ##

### 核心思想 ###

通过定时任务扫描，执行业务逻辑。

--------------------

### Demo Code ###

参考实现如下：

<dependency>
            <groupId>org.quartz-scheduler</groupId>
            <artifactId>quartz</artifactId>
            <version>2.2.2</version>
        </dependency>

import org.quartz.JobBuilder;
    import org.quartz.JobDetail;
    import org.quartz.Scheduler;
    import org.quartz.SchedulerException;
    import org.quartz.SchedulerFactory;
    import org.quartz.SimpleScheduleBuilder;
    import org.quartz.Trigger;
    import org.quartz.TriggerBuilder;
    import org.quartz.impl.StdSchedulerFactory;
    import org.quartz.Job;
    import org.quartz.JobExecutionContext;
    import org.quartz.JobExecutionException;
     
    public class MyJob implements Job { 
        public void execute(JobExecutionContext context)
                throws JobExecutionException { 
            System.out.println("模拟扫描任务。。。。。");
        }
     
        public static void main(String[] args) throws Exception { 
            // 创建任务
            JobDetail jobDetail = JobBuilder.newJob(MyJob.class)
                    .withIdentity("job1", "group1").build();
            // 创建触发器 每3秒钟执行一次
            Trigger trigger = TriggerBuilder
                    .newTrigger()
                    .withIdentity("trigger1", "group3")
                    .withSchedule(
                            SimpleScheduleBuilder.simpleSchedule()
                                    .withIntervalInSeconds(3).repeatForever())
                    .build();
            Scheduler scheduler = new StdSchedulerFactory().getScheduler();
            // 将任务及其触发器放入调度器
            scheduler.scheduleJob(jobDetail, trigger);
            // 调度器开始调度任务
            scheduler.start();
        }
    }

--------------------

### 优缺点 ###

优点: 简单 （好像也没有其他的优点了 哈哈哈 ）

![在这里插入图片描述][22f49d1e64354057a7341f7fe0f3fa14.png]

缺点:

*  (1)占用资源，对服务器内存消耗大
 *  (2)存在延迟，比如你每隔n分钟扫描一次，那最坏的延迟时间就是n分钟
 *  (3)如果表的数据量较大，每隔几分钟这样扫描一次，性能堪忧，DB压力较大

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_10_color_FFFFFF_t_70_g_se_x_16]

--------------------

## JDK的Delay Queue ##

### 核心思想 ###

利用JDK自带的DelayQueue来实现， 无界阻塞队列，该队列只有在延迟期满的时候才能从中获取元素，放入DelayQueue中的对象，必须实现Delayed接口。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16]

*  poll():获取并移除队列的超时元素，没有则返回空
 *  take():获取并移除队列的超时元素，如果没有则wait当前线程，直到有元素满足超时条件，返回结果。

--------------------

### Demo Code ###

import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/9/2 22:50 * @mark: show me the code , change the world */
    public class TicketDelay implements Delayed { 
    
        private String ticketId;
        private long timeout;
    
        public TicketDelay(String ticketId, long timeout) { 
            this.ticketId= ticketId;
            this.timeout = timeout + System.nanoTime();
        }
    
        @Override
        public int compareTo(Delayed other) { 
            if (other == this) { 
                return 0;
            }
    
            TicketDelay t = (TicketDelay) other;
            long d = (getDelay(TimeUnit.NANOSECONDS) - t.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
        }
    
        /** * 返回距离自定义的超时时间还有多少 * @param unit * @return */
        @Override
        public long getDelay(TimeUnit unit) { 
            return unit.convert(timeout - System.nanoTime(),TimeUnit.NANOSECONDS);
        }
    
        void doSomething() { 
            System.out.println(ticketId+" is deleted");
        }
    }

import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.TimeUnit;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/9/2 22:51 * @mark: show me the code , change the world */
    public class DelayQueueDemo { 
    
        public static void main(String[] args) { 
    
            // 模拟数据
            List<String> list = new ArrayList<>();
            list.add("Ticket-1");
            list.add("Ticket-2");
            list.add("Ticket-3");
            list.add("Ticket-4");
            list.add("Ticket-5");
    
            // 延时队列
            DelayQueue<TicketDelay> queue = new DelayQueue<>();
    
    
            for (int i = 0; i < 5; i++) { 
                long start = System.currentTimeMillis();
                //延迟2秒取出
                queue.put(new TicketDelay(list.get(i), TimeUnit.NANOSECONDS.convert(2, TimeUnit.SECONDS)));
                System.out.println("biubiubiu ~ " + (System.currentTimeMillis() - start) + " MilliSeconds ");
    
                try { 
                    queue.take().doSomething();
                    System.out.println("biubiubiu " + (System.currentTimeMillis() - start) + " MilliSeconds 取到了数据，开始后执行业务操作");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println("===========================\n" );
            }
        }
    }

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 1]

### 优缺点 ###

优点:

*  效率高,任务触发时间延迟低  
    ![在这里插入图片描述][2d7eebd42fdd4844be855448e8018101.png]

缺点:

*  服务器重启后，数据全部消失，怕宕机
 *  集群扩展相当麻烦
 *  因为内存条件限制的原因，如数据太多，那么很容易就出现OOM异常
 *  代码复杂度较高

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_15_color_FFFFFF_t_70_g_se_x_16]

--------------------

## 时间轮算法 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 2]

[每日一博 - 使用环形队列实现高效的延时消息][-]

[延时消息之时间轮][Link 1]

--------------------

### 核心思想 ###

其实本质上它就是一个环形的数组，如图所示，假设我们创建了一个长度为 8 的时间轮。

task0 = 当我们需要新建一个 5s 延时消息，则只需要将它放到下标为 5 的那个槽中。

task1 = 而如果是一个 10s 的延时消息，则需要将它放到下标为 2 的槽中，但同时需要记录它所对应的圈数，不然就和 2 秒的延时消息重复了。

task2= 当创建一个 21s 的延时消息时，它所在的位置就和 task0 相同了，都在下标为 5 的槽中，所以为了区别需要为他加上圈数为 2。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 3]

当我们需要取出延时消息时，只需要每秒往下移动这个指针，然后取出该位置的所有任务即可。

当然取出任务之前还得判断圈数是否为 0 ，不为 0 时说明该任务还得再轮几圈，同时需要将圈数 -1 。

这样就可避免轮询所有的任务，不过如果时间轮的槽比较少，导致某一个槽上的任务非常多那效率也比较低，这就和 HashMap 的 hash 冲突是一样的。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_10_color_FFFFFF_t_70_g_se_x_16 1]

--------------------

时间轮算法可以类比于时钟， （指针）按某一个方向按固定频率轮动，每一次跳动称为一个 tick。这样可以看出定时轮由个3个重要的属性参数，ticksPerWheel（一轮的tick数），tickDuration（一个tick的持续时间）以及 timeUnit（时间单位）。

例如当ticksPerWheel=60，tickDuration=1，timeUnit=秒，这就和现实中的始终的秒针走动完全类似了。

如果当前指针指在1上面，我有一个任务需要4秒以后执行，那么这个执行的线程回调或者消息将会被放在5上。那如果需要在20秒之后执行怎么办，由于这个环形结构槽数只到8，如果要20秒，指针需要多转2圈。位置是在2圈之后的5上面（20 % 8 + 1）

--------------------

### Demo Code ###

我们用Netty的HashedWheelTimer来实现

<dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.24.Final</version>
            </dependency>

import io.netty.util.HashedWheelTimer;
    import io.netty.util.Timeout;
    import io.netty.util.Timer;
    import io.netty.util.TimerTask;
    
    import java.util.concurrent.TimeUnit;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/9/2 23:30 * @mark: show me the code , change the world */
    public class HashedWheelTimerTest { 
    
        static class MyTimerTask implements TimerTask { 
            boolean flag;
    
            public MyTimerTask(boolean flag) { 
                this.flag = flag;
            }
    
            @Override
            public void run(Timeout timeout) throws Exception { 
                System.out.println("要去执行业务啦....");
                this.flag = false;
            }
        }
    
        public static void main(String[] argv) { 
            MyTimerTask timerTask = new MyTimerTask(true);
            Timer timer = new HashedWheelTimer();
            timer.newTimeout(timerTask, 5, TimeUnit.SECONDS);
            int i = 1;
            while (timerTask.flag) { 
                try { 
                    Thread.sleep(1000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println(i + "秒过去了");
                i++;
            }
        }
    }

### 优缺点 ###

优点

*  效率高,任务触发时间延迟时间比delayQueue低，代码复杂度比delayQueue低。

缺点:

*  服务器重启后，数据全部消失，怕宕机
 *  集群扩展相当麻烦
 *  因为内存条件限制的原因，比如数据太多，那么很容易就出现OOM异常

--------------------

## Redis缓存（zset） ##

### 核心思想 ###

利用redis的zset,zset是一个有序集合，每一个元素(member)都关联了一个score,通过score排序来取集合中的值

# 添加单个元素
     
    redis> ZADD page_rank 10 google.com
    (integer) 1
     
     
    # 添加多个元素
     
    redis> ZADD page_rank 9 baidu.com 8 bing.com
    (integer) 2
     
    redis> ZRANGE page_rank 0 -1 WITHSCORES
    1) "bing.com"
    2) "8"
    3) "baidu.com"
    4) "9"
    5) "google.com"
    6) "10"
     
    # 查询元素的score值
    redis> ZSCORE page_rank bing.com
    "8"
     
    # 移除单个元素
     
    redis> ZREM page_rank google.com
    (integer) 1
     
    redis> ZRANGE page_rank 0 -1 WITHSCORES
    1) "bing.com"
    2) "8"
    3) "baidu.com"
    4) "9"

那么如何实现呢？我们**将订单超时时间戳与订单号分别设置为score和member,系统扫描第一个元素判断是否超时**

### Demo Code ###

import java.util.Calendar;
    import java.util.Set;
    
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPool;
    import redis.clients.jedis.Tuple;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/9/4 21:04 * @mark: show me the code , change the world */
    public class RedisDelayQueue { 
    
        private static final String ADDR = "127.0.0.1";
        private static final int PORT = 6379;
        private static JedisPool jedisPool = new JedisPool(ADDR, PORT);
    
        public static Jedis getJedis() { 
            return jedisPool.getResource();
        }
    
        /** * 生产者,生成5个订单放进去 */
        public void productionDelayMessage() { 
            for (int i = 0; i < 5; i++) { 
                //延迟3秒
                Calendar cal1 = Calendar.getInstance();
                cal1.add(Calendar.SECOND, 3);
                int second3later = (int) (cal1.getTimeInMillis() / 1000);
                RedisDelayQueue.getJedis().zadd("OrderId", second3later, "ARTISAN_ID_" + i);
                System.out.println(System.currentTimeMillis() + "ms:redis生成了订单：订单ID为" + "ARTISAN_ID_" + i);
            }
        }
    
        /** * 消费者，取订单 */
        public void consumerDelayMessage() { 
            Jedis jedis = RedisDelayQueue.getJedis();
            while (true) { 
                Set<Tuple> items = jedis.zrangeWithScores("OrderId", 0, 1);
                if (items == null || items.isEmpty()) { 
                    System.out.println("当前没有等待的任务");
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    continue;
                }
                int score = (int) ((Tuple) items.toArray()[0]).getScore();
                Calendar cal = Calendar.getInstance();
                int nowSecond = (int) (cal.getTimeInMillis() / 1000);
                if (nowSecond >= score) { 
                    String orderId = ((Tuple) items.toArray()[0]).getElement();
                    jedis.zrem("OrderId", orderId);
                    System.out.println(System.currentTimeMillis() + "ms:redis消费了一个任务：消费的订单Id为" + orderId);
     
                }
            }
        }
    
        public static void main(String[] args) { 
            RedisDelayQueue appTest = new RedisDelayQueue();
            appTest.productionDelayMessage();
            appTest.consumerDelayMessage();
        }
    }

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 4]

上面的代码有个硬伤：在高并发条件下，多消费者会取到同一个订单号。

import java.util.concurrent.CountDownLatch;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/9/4 21:21 * @mark: show me the code , change the world */
    public class MTest { 
    
        private static final int threadNum = 10;
    
        private static CountDownLatch cdl = new CountDownLatch(threadNum);
    
        static class DelayMessage implements Runnable { 
            @Override
            public void run() { 
                try { 
                    cdl.await();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                RedisDelayQueue appTest = new RedisDelayQueue();
                appTest.consumerDelayMessage();
            }
        }
    
        public static void main(String[] args) { 
            RedisDelayQueue appTest = new RedisDelayQueue();
            appTest.productionDelayMessage();
            for (int i = 0; i < threadNum; i++) { 
                new Thread(new DelayMessage()).start();
                cdl.countDown();
            }
        }
    }

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 5]

显然，出现了多个线程消费同一个资源的情况

解决方案

*  (1)用分布式锁，但是用分布式锁，性能下降了，不推荐
 *  (2)对ZREM的返回值进行判断，只有大于0的时候，才消费数据，于是将`consumerDelayMessage()`方法里的

if(nowSecond >= score){ 
    	    String orderId = ((Tuple)items.toArray()[0]).getElement();
    	    jedis.zrem("OrderId", orderId);
    	    System.out.println(System.currentTimeMillis()+"ms:redis消费了一个任务：消费的订单OrderId为"+orderId);
    	}

修改为

if(nowSecond >= score){ 
    	    String orderId = ((Tuple)items.toArray()[0]).getElement();
    	    Long num = jedis.zrem("OrderId", orderId);
    	    if( num != null && num>0){ 
    	        System.out.println(System.currentTimeMillis()+"ms:redis消费了一个任务：消费的订单OrderId为"+orderId);
    	    }
    	}

--------------------

## Redis缓存（Keyspace Notifications） ##

### 核心思想 ###

该方案使用redis的Keyspace Notifications，中文翻译就是键空间机制，就是利用该机制可以在key失效之后，提供一个回调，实际上是redis会给客户端发送一个消息。 redis版本2.8以上。

在redis.conf中，加入一条配置

notify-keyspace-events Ex

import redis.clients.jedis.JedisPool;
    import redis.clients.jedis.JedisPubSub;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/9/4 21:38 * @mark: show me the code , change the world */
    public class RedisTest { 
    
        private static final String ADDR = "127.0.0.1";
        private static final int PORT = 6379;
        private static JedisPool jedis = new JedisPool(ADDR, PORT);
        private static RedisSub sub = new RedisSub();
    
        public static void init() { 
            new Thread(() -> jedis.getResource().subscribe(sub, "__keyevent@0__:expired")).start();
        }
    
        public static void main(String[] args) throws InterruptedException { 
            init();
            for (int i = 0; i < 10; i++) { 
                String orderId = "OID000000" + i;
                jedis.getResource().setex(orderId, 3, orderId);
                System.out.println(System.currentTimeMillis() + "ms:" + orderId + "订单生成");
            }
        }
    
        static class RedisSub extends JedisPubSub { 
            @Override
            public void onMessage(String channel, String message) { 
                System.out.println(System.currentTimeMillis() + "ms:" + message + "订单取消");
            }
        }
    }

Redis的发布/订阅目前是即发即弃(fire and forget)模式的，因此无法实现事件的可靠通知。也就是说，如果发布/订阅的客户端断链之后又重连，则在客户端断链期间的所有事件都丢失了。  
因此，Keyspace Notifications不是太推荐。当然，如果你对可靠性要求不高，可以使用。

--------------------

## Redisson 延时队列 ##

## 使用消息队列 ##

![TODO][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70]: /images/20220829/4bb6e662591a4271b58300f1d99b3693.png
[-]: https://blog.csdn.net/yangshangwei/article/details/119258887?spm=1001.2014.3001.5501
[7b841a624f6649d0a3d1330ef0527c47.png]: /images/20220829/c89300285c914f94a216786e29fc7c48.png
[22f49d1e64354057a7341f7fe0f3fa14.png]: /images/20220829/8cf714a7bdfa4d09b309b773b843865c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_10_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/0df033dbda094895a533a5a0895aecf0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/2b5bdabc07e14ef39c512a41070be438.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/4b5cfb7ba35f49a589f9d799d16f2be7.png
[2d7eebd42fdd4844be855448e8018101.png]: /images/20220829/898580731c62471f8f292244b012a935.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_15_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/df67cb6fa7db4787a817424dd34ee100.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/b9a29615fa0e477ba368d99c024502db.png
[Link 1]: https://crossoverjie.top/2019/09/27/algorithm/time%20wheel/
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/d3711678798b45eab3cdfe141762169c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_10_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/160faf568363493fbf58f3d06515856f.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220829/a32dafe3685647bf9bea129891d69ec2.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP5bCP5bel5Yyg_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220829/415fd822790349ebb6c2a78a90cdf709.png
[TODO]: /images/20220829/8afb82bcd343435c89fb9707f3024224.png