Java處理延時任務的常用幾種解決方案

更新時間：2022年05月31日 16:04:12 作者：Monster_起飛

本文主要介紹了Java處理延時任務的常用幾種解決方案，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

前言

項目中經(jīng)常會遇到如下的需求：

創(chuàng)建訂單30分鐘未支付，訂單自動取消。
訂單支付成功后，1分鐘后給用戶發(fā)送短信，提醒用戶評價。
…

針對延時任務需求，我們可以采用如下的解決方案：

數(shù)據(jù)庫輪詢

原理

通過一個線程定時的掃描數(shù)據(jù)庫當天創(chuàng)建的訂單,根據(jù)訂單的創(chuàng)建時間來判斷訂單是否超時，針對超時訂單進行相關(guān)的更新操作。
實現(xiàn)技術(shù)
采用Spring Boot結(jié)合quartz來實現(xiàn)，具體的實現(xiàn)可以參考之前的文章。

優(yōu)缺點

優(yōu)點：

此方案比較簡單,且quartz也支持集群操作。

缺點：

系統(tǒng)訂單數(shù)據(jù)量比較大，每個幾分鐘輪詢數(shù)據(jù)庫，對服務器和數(shù)據(jù)庫的內(nèi)存消耗比較大。
存在延遲，即使1分鐘掃描一次數(shù)據(jù)庫，也會存在1分鐘的延遲。

Java延遲隊列

原理

采用JDK自帶的DelayQueue來實現(xiàn)，這是一個無界阻塞隊列，該隊列只有在延遲期滿的時候才能從中獲取元素，放入DelayQueue中的對象。
實現(xiàn)技術(shù)
使用JDK的DelayQueue隊列進行相關(guān)操作即可。

優(yōu)缺點

優(yōu)點：

此方案是基于內(nèi)存操作所以效率高,任務觸發(fā)時間延遲低.

缺點：

消息隊列的信息都存放在內(nèi)存中，一旦服務器重啟，則數(shù)據(jù)全部消失
無法進行集群用擴展
由于本機內(nèi)存有限，一旦訂單數(shù)據(jù)量過大，很容易出現(xiàn)OOM異常。

Reids監(jiān)聽失效key

原理

該方案使用Redis的Keyspace Notifications，利用key失效的提供的回調(diào)機制，處理相關(guān)的業(yè)務實現(xiàn)

實現(xiàn)技術(shù)

基于reids的方案，實現(xiàn)MessageListener接口。

實現(xiàn)步驟

修改Redis配置文件
打開redis.conf 文件,搜索 “notify-keyspace-events”找到原本的notify-keyspace-events " "，修改為 “notify-keyspace-events Ex”,至此Redis 就支持Key過期事件的監(jiān)聽。

創(chuàng)建監(jiān)聽類，實現(xiàn)MessageListener接口

@Component
public class RedisKeyExpirationListener implements MessageListener
{
    private static final Logger logger = LoggerFactory.getLogger(RedisKeyExpirationListener.class);
    
    public static final String KEY_PREX = "test::order:queue";
    
    @Override
    public void onMessage(Message message, byte[] pattern)
    {
        try
        {
            String expiredKey = message.toString();

            // 通過key來判斷
            if(!expiredKey.contains(KEY_PREX))
            {
                return;
            }
            
            //滿足條件處理具體的業(yè)務邏輯
        }
        catch (Exception e)
        {
            logger.error("失效事件失敗",e);
        }
    }
}

優(yōu)缺點

優(yōu)點：

基于Redis實現(xiàn)簡單

缺點：

客戶端斷開后重連會導致所有事件丟失。
高并發(fā)場景下，存在大量的失效key場景會導出失效時間存在延遲。
此方案針對業(yè)務量較少且可靠性要求不高的場景使用。

RocketMq延遲消息

實現(xiàn)原理

基于RocketMQ設置消息的等級，發(fā)送延遲消息，RocketMQ延時消息會暫存在名為SCHEDULE_TOPIC_XXXX的Topic中，并根據(jù)delayTimeLevel存入特定的queue，queueId = delayTimeLevel – 1，即一個queue只存相同延遲的消息，保證具有相同發(fā)送延遲的消息能夠順序消費。broker會調(diào)度地消費SCHEDULE_TOPIC_XXXX，將消息寫入真實的topic。
其具體步驟如下：

修改消息Topic名稱和隊列信息
轉(zhuǎn)發(fā)消息到延遲主題SCHEDULE_TOPIC_XXXX的CosumeQueue中
延遲服務消費SCHEDULE_TOPIC_XXXX消息
將信息重新存儲到CommitLog中
將消息投遞到目標Topic中
消費者消費目標topic中的數(shù)據(jù)。

/**
    * 發(fā)送延遲消息
    * @param topic
    * @param msg
    */
   public void sendDelayMessage(String topic,Object msg)
   {
      Message msgMessage =new Message();
      //設置消息等級
      msgMessage.setDelayTimeLevel(2);
     rocketMQTemplate.convertAndSend(topic, msg);
   }

注意：RocketMQ延時消息的延遲時長不支持隨意時長的延遲，是通過特定的延遲等級來指定的。默認支持18個等級的延遲消息，延時等級定義在RocketMQ服務端的MessageStoreConfig類中的如下變量中：