前言

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

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

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

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

原理

通過一個線程定時的掃描數(shù)據(jù)庫當(dāng)天創(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ù)庫，對服務(wù)器和數(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)存操作所以效率高,任務(wù)觸發(fā)時間延遲低.

缺點：

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

Reids監(jiān)聽失效key

原理

該方案使用Redis的Keyspace Notifications，利用key失效的提供的回調(diào)機制，處理相關(guān)的業(yè)務(wù)實現(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è)務(wù)邏輯
        }
        catch (Exception e)
        {
            logger.error("失效事件失敗",e);
        }
    }
}

優(yōu)缺點

優(yōu)點：

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

缺點：

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

RocketMq延遲消息

實現(xiàn)原理

基于RocketMQ設(shè)置消息的等級，發(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中
• 延遲服務(wù)消費SCHEDULE_TOPIC_XXXX消息
• 將信息重新存儲到CommitLog中
• 將消息投遞到目標(biāo)Topic中
• 消費者消費目標(biāo)topic中的數(shù)據(jù)。

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

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

例如指定的延時等級為2，則表示延遲時長為5s，即延遲等級是從1開始計數(shù)的。

優(yōu)缺點

優(yōu)點：

支持高并發(fā)場景消息處理.

缺點：

• 引入額外的消息隊列，增加項目的維護和復(fù)雜度。
• 支持固定時長的消息延遲，針對任意時長的消息延遲需要進行擴展。

總結(jié)

本文講解了針對延時任務(wù)的處理的幾種方案和相關(guān)的優(yōu)缺點，針對不同的業(yè)務(wù)場景，選擇合適的解決方案。更多相關(guān)Java 延時任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論