硬核！史上最全的延遲任務實現方式彙總！附代碼（強烈推薦）

時間  2020-05-10
標籤 硬核 史上 延遲 任務 實現 方式 彙總 代碼 推薦 简体版
原文   原文鏈接

延遲任務要怎麼實現？html

話很少說，直接進入今天的主題，本文的主要內容以下圖所示：java

8d6bd6f385824d26b966ed9ac064482e


什麼是延遲任務？

顧明思議，咱們把須要延遲執行的任務叫作延遲任務。程序員

延遲任務的使用場景有如下這些：面試

  1. 紅包 24 小時未被查收，須要延遲執退還業務；redis

  2. 每月帳單日，須要給用戶發送當月的對帳單；spring

  3. 訂單下單以後 30 分鐘後，用戶若是沒有付錢，系統須要自動取消訂單。編程

等事件都須要使用延遲任務。服務器

延遲任務實現思路分析

延遲任務實現的關鍵是在某個時間節點執行某個任務。基於這個信息咱們能夠想到實現延遲任務的手段有如下兩個：網絡

  1. 本身手寫一個「死循環」一直判斷當前時間節點有沒有要執行的任務；數據結構

  2. 藉助 JDK 或者第三方提供的工具類來實現延遲任務。

而經過 JDK 實現延遲任務咱們能想到的關鍵詞是：DelayQueue、ScheduledExecutorService，而第三方提供的延遲任務執行方法就有不少了，例如：Redis、Netty、MQ 等手段。

延遲任務實現

下面咱們將結合代碼來說解每種延遲任務的具體實現。

1.無限循環實現延遲任務

此方式咱們須要開啓一個無限循環一直掃描任務，而後使用一個 Map 集合用來存儲任務和延遲執行的時間，實現代碼以下：

import java.time.Instant;import java.time.LocalDateTime;import java.util.HashMap;import java.util.Iterator;import java.util.Map;/**
 * 延遲任務執行方法彙總
 */public class DelayTaskExample {
    // 存放定時任務    private static Map<String, Long> _TaskMap = new HashMap<>();

    public static void main(String[] args) {
        System.out.println("程序啓動時間：" + LocalDateTime.now());
        // 添加定時任務        _TaskMap.put("task-1", Instant.now().plusSeconds(3).toEpochMilli()); // 延遲 3s        // 調用無限循環實現延遲任務        loopTask();
    }

    /**
     * 無限循環實現延遲任務
     */    public static void loopTask() {
        Long itemLong = 0L;
        while (true) {
            Iterator it = _TaskMap.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry entry = (Map.Entry) it.next();
                itemLong = (Long) entry.getValue();
                // 有任務須要執行                if (Instant.now().toEpochMilli() >= itemLong) {
                    // 延遲任務，業務邏輯執行                    System.out.println("執行任務：" + entry.getKey() +
                            " ，執行時間：" + LocalDateTime.now());
                    // 刪除任務                    _TaskMap.remove(entry.getKey());
                }
            }
        }
    }
}

以上程序執行的結果爲：

程序啓動時間：2020-04-12T18:51:28.188

執行任務：task-1 ，執行時間：2020-04-12T18:51:31.189

能夠看出任務延遲了 3s 鍾執行了，符合咱們的預期。

2.Java API 實現延遲任務

Java API 提供了兩種實現延遲任務的方法：DelayQueue 和 ScheduledExecutorService。

① ScheduledExecutorService 實現延遲任務

咱們可使用 ScheduledExecutorService 來以固定的頻率一直執行任務，實現代碼以下：

81e51857029d446ebff47f5b5ab7a63d


以上程序執行的結果爲：

程序啓動時間：2020-04-12T21:28:10.416

執行任務 ，執行時間：2020-04-12T21:28:12.421

執行任務 ，執行時間：2020-04-12T21:28:14.422

......

能夠看出使用 ScheduledExecutorService#scheduleWithFixedDelay(...) 方法以後，會以某個頻率一直循環執行延遲任務。

② DelayQueue 實現延遲任務

DelayQueue 是一個支持延時獲取元素的***阻塞隊列，隊列中的元素必須實現 Delayed 接口，並重寫 getDelay(TimeUnit) 和 compareTo(Delayed) 方法，DelayQueue 實現延遲隊列的完整代碼以下：

public class DelayTest {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue delayQueue = new DelayQueue();
        // 添加延遲任務        delayQueue.put(new DelayElement(1000));
        delayQueue.put(new DelayElement(3000));
        delayQueue.put(new DelayElement(5000));
        System.out.println("開始時間：" +  DateFormat.getDateTimeInstance().format(new Date()));
        while (!delayQueue.isEmpty()){
            // 執行延遲任務            System.out.println(delayQueue.take());
        }
        System.out.println("結束時間：" +  DateFormat.getDateTimeInstance().format(new Date()));
    }

    static class DelayElement implements Delayed {
        // 延遲截止時間（單面：毫秒）        long delayTime = System.currentTimeMillis();
        public DelayElement(long delayTime) {
            this.delayTime = (this.delayTime + delayTime);
        }
        @Override        // 獲取剩餘時間        public long getDelay(TimeUnit unit) {
            return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        @Override        // 隊列裏元素的排序依據        public int compareTo(Delayed o) {
            if (this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) {
                return 1;
            } else if (this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS)) {
                return -1;
            } else {
                return 0;
            }
        }
        @Override        public String toString() {
            return DateFormat.getDateTimeInstance().format(new Date(delayTime));
        }
    }
}

以上程序執行的結果爲：

開始時間：2020-4-12 20:40:38

2020-4-12 20:40:39

2020-4-12 20:40:41

2020-4-12 20:40:43

結束時間：2020-4-12 20:40:43

3.Redis 實現延遲任務

使用 Redis 實現延遲任務的方法大致可分爲兩類：經過 zset 數據判斷的方式，和經過鍵空間通知的方式。

① 經過數據判斷的方式

咱們藉助 zset 數據類型，把延遲任務存儲在此數據集合中，而後在開啓一個無線循環查詢當前時間的全部任務進行消費，實現代碼以下（須要藉助 Jedis 框架）：

import redis.clients.jedis.Jedis;import utils.JedisUtils;import java.time.Instant;import java.util.Set;public class DelayQueueExample {
    // zset key    private static final String _KEY = "myDelayQueue";
    
    public static void main(String[] args) throws InterruptedException {
        Jedis jedis = JedisUtils.getJedis();
        // 延遲 30s 執行（30s 後的時間）        long delayTime = Instant.now().plusSeconds(30).getEpochSecond();
        jedis.zadd(_KEY, delayTime, "order_1");
        // 繼續添加測試數據        jedis.zadd(_KEY, Instant.now().plusSeconds(2).getEpochSecond(), "order_2");
        jedis.zadd(_KEY, Instant.now().plusSeconds(2).getEpochSecond(), "order_3");
        jedis.zadd(_KEY, Instant.now().plusSeconds(7).getEpochSecond(), "order_4");
        jedis.zadd(_KEY, Instant.now().plusSeconds(10).getEpochSecond(), "order_5");
        // 開啓延遲隊列        doDelayQueue(jedis);
    }

    /**
     * 延遲隊列消費
     * @param jedis Redis 客戶端
     */    public static void doDelayQueue(Jedis jedis) throws InterruptedException {
        while (true) {
            // 當前時間            Instant nowInstant = Instant.now();
            long lastSecond = nowInstant.plusSeconds(-1).getEpochSecond(); // 上一秒時間            long nowSecond = nowInstant.getEpochSecond();
            // 查詢當前時間的全部任務            Set<String> data = jedis.zrangeByScore(_KEY, lastSecond, nowSecond);
            for (String item : data) {
                // 消費任務                System.out.println("消費：" + item);
            }
            // 刪除已經執行的任務            jedis.zremrangeByScore(_KEY, lastSecond, nowSecond);
            Thread.sleep(1000); // 每秒輪詢一次        }
    }
}

② 經過鍵空間通知

默認狀況下 Redis 服務器端是不開啓鍵空間通知的，須要咱們經過 config set notify-keyspace-events Ex 的命令手動開啓，開啓鍵空間通知後，咱們就能夠拿到每一個鍵值過時的事件，咱們利用這個機制實現了給每一個人開啓一個定時任務的功能，實現代碼以下：

e1cf9a6a6eee4b4d9e3582ee29c5ddc6


4.Netty 實現延遲任務

Netty 是由 JBOSS 提供的一個 Java 開源框架，它是一個基於 NIO 的客戶、服務器端的編程框架，使用 Netty 能夠確保你快速和簡單的開發出一個網絡應用，例如實現了某種協議的客戶、服務端應用。Netty 至關於簡化和流線化了網絡應用的編程開發過程，例如：基於 TCP 和 UDP 的 socket 服務開發。

可使用 Netty 提供的工具類 HashedWheelTimer 來實現延遲任務，實現代碼以下。

首先在項目中添加 Netty 引用，配置以下：

9fad78d276244fe9a79e7f43056a5639


Netty 實現的完整代碼以下：

0432c49b38154ecf8c14c873393a0040


以上程序執行的結果爲：

程序啓動時間：2020-04-13T10:16:23.033

執行任務 ，執行時間：2020-04-13T10:16:26.118

HashedWheelTimer 是使用定時輪實現的，定時輪其實就是一種環型的數據結構，能夠把它想象成一個時鐘，分紅了許多格子，每一個格子表明必定的時間，在這個格子上用一個鏈表來保存要執行的超時任務，同時有一個指針一格一格的走，走到那個格子時就執行格子對應的延遲任務，以下圖所示：

4b9fd55b865049ff8849aa3891799d96


(圖片來源於網絡)

以上的圖片能夠理解爲，時間輪大小爲 8，某個時間轉一格（例如 1s），每格指向一個鏈表，保存着待執行的任務。

5.MQ 實現延遲任務

若是專門開啓一個 MQ 中間件來執行延遲任務，就有點殺雞用宰牛刀般的奢侈了，不過已經有了 MQ 環境的話，用它來實現延遲任務的話，仍是可取的。

幾乎全部的 MQ 中間件均可以實現延遲任務，在這裏更準確的叫法應該叫延隊列。本文就使用 RabbitMQ 爲例，來看它是如何實現延遲任務的。

RabbitMQ 實現延遲隊列的方式有兩種：

  • 經過消息過時後進入死信交換器，再由交換器轉發到延遲消費隊列，實現延遲功能；

  • 使用 rabbitmq-delayed-message-exchange 插件實現延遲功能。

注意：延遲插件 rabbitmq-delayed-message-exchange 是在 RabbitMQ 3.5.7 及以上的版本才支持的，依賴 Erlang/OPT 18.0 及以上運行環境。

因爲使用死信交換器比較麻煩，因此推薦使用第二種實現方式 rabbitmq-delayed-message-exchange 插件的方式實現延遲隊列的功能。

首先，咱們須要下載並安裝 rabbitmq-delayed-message-exchange 插件，下載地址：http://www.rabbitmq.com/community-plugins.html

選擇相應的對應的版本進行下載，而後拷貝到 RabbitMQ 服務器目錄，使用命令rabbitmq-plugins enable rabbitmq_delayed_message_exchange 開啓插件，在使用命令 rabbitmq-plugins list 查詢安裝的全部插件，安裝成功以下圖所示：

1654e12e9b4740af9e938d125a19ae55


最後重啓 RabbitMQ 服務，使插件生效。

首先，咱們先要配置消息隊列，實現代碼以下：

import com.example.rabbitmq.mq.DirectConfig;import org.springframework.amqp.core.*;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import java.util.HashMap;import java.util.Map;@Configurationpublic class DelayedConfig {
    final static String QUEUE_NAME = "delayed.goods.order";
    final static String EXCHANGE_NAME = "delayedec";
    @Bean    public Queue queue() {
        return new Queue(DelayedConfig.QUEUE_NAME);
    }

    // 配置默認的交換機    @Bean    CustomExchange customExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        //參數二爲類型：必須是x-delayed-message        return new CustomExchange(DelayedConfig.EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }
    // 綁定隊列到交換器    @Bean    Binding binding(Queue queue, CustomExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(DelayedConfig.QUEUE_NAME).noargs();
    }
}

而後添加增長消息的代碼，具體實現以下：

import org.springframework.amqp.AmqpException;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.amqp.core.Message;import org.springframework.amqp.core.MessagePostProcessor;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.text.SimpleDateFormat;import java.util.Date;@Componentpublic class DelayedSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(String msg) {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("發送時間：" + sf.format(new Date()));

        rabbitTemplate.convertAndSend(DelayedConfig.EXCHANGE_NAME, DelayedConfig.QUEUE_NAME, msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setHeader("x-delay", 3000);
                return message;
            }
        });
    }
}

再添加消費消息的代碼：

441c8cf8237c4b3c9534f279f3ee76a0


最後，咱們使用代碼測試一下：

b41d29695fc94f28b22ce9558f7e437a


以上程序的執行結果以下：

發送時間：2020-04-13 20:47:51

接收時間:2020-04-13 20:47:54

消息內容：Hi Admin.

從結果能夠看出，以上程序執行符合延遲任務的實現預期。

6.使用 Spring 定時任務

若是你使用的是 Spring 或 SpringBoot 的項目的話，可使用藉助 Scheduled 來實現，本文將使用 SpringBoot 項目來演示 Scheduled 的實現，實現咱們須要聲明開啓 Scheduled，實現代碼以下：

10a53bf4464842eab0d5b6eeab6bd31a


而後添加延遲任務，實現代碼以下：

bef0e6a39865499dac6183c11ff8a150


此時當咱們啓動項目以後就能夠看到任務以延遲了 2s 的形式一直循環執行，結果以下：

任務執行，時間：2020-04-13T14:07:53.349

任務執行，時間：2020-04-13T14:07:55.350

任務執行，時間：2020-04-13T14:07:57.351

...

咱們也可使用 Corn 表達式來定義任務執行的頻率，例如使用 @Scheduled(cron = "0/4 * * * * ?") 。

7.Quartz 實現延遲任務

Quartz 是一款功能強大的任務調度器，能夠實現較爲複雜的調度功能，它還支持分佈式的任務調度。

咱們使用 Quartz 來實現一個延遲任務，首先定義一個執行任務代碼以下：

55d142ab541c46c285bb173357f27498


在定義一個 JobDetail 和 Trigger 實現代碼以下：

4b098c784f9e451e87a56b6da546e78d


最後在 SpringBoot 項目啓動以後開啓延遲任務，實現代碼以下：

24b49c2f5fbd4b17b7d8115bd1d3fd3a


以上程序的執行結果以下：

2020-04-13 19:02:12.331 INFO 17768 --- [ restartedMain] com.example.demo.DemoApplication : Started DemoApplication in 1.815 seconds (JVM running for 3.088)

任務執行，時間：2020-04-13T19:02:15.019

從結果能夠看出在項目啓動 3s 以後執行了延遲任務。

總結

本文講了延遲任務的使用場景，以及延遲任務的 10 種實現方式：

  1. 手動無線循環；

  2. ScheduledExecutorService；

  3. DelayQueue；

  4. Redis zset 數據判斷的方式；

  5. Redis 鍵空間通知的方式；

  6. Netty 提供的 HashedWheelTimer 工具類；

  7. RabbitMQ 死信隊列；

  8. RabbitMQ 延遲消息插件 rabbitmq-delayed-message-exchange；

  9. Spring Scheduled；

  10. Quartz。

推薦閱讀：

馬士兵親述美團七道面試題，深刻JVM底層原理（想去美團、大廠的小夥伴建議看看）

年薪百萬架構師馬士兵親述美團七面，深刻解析JVM骨髓_嗶哩嗶哩 (゜-゜)つロ 乾杯~-bilibiliwww.bilibili.com馬士兵親述美團面試七道題，不想進大廠的程序員不是好程序員_嗶哩嗶哩 (゜-゜)つロ 乾杯~-bilibiliwww.bilibili.com

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程