前言

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

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

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

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

原理

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

優(yōu)缺點(diǎn)

優(yōu)點(diǎn)：

此方案比較簡(jiǎn)單,且quartz也支持集群操作。

缺點(diǎn)：

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

Java延遲隊(duì)列

原理

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

優(yōu)缺點(diǎn)

優(yōu)點(diǎn)：

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

缺點(diǎn)：

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

Reids監(jiān)聽失效key

原理

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

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

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

實(shí)現(xiàn)步驟

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

創(chuàng)建監(jiān)聽類，實(shí)現(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)缺點(diǎn)

優(yōu)點(diǎn)：

基于Redis實(shí)現(xiàn)簡(jiǎn)單

缺點(diǎn)：

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

RocketMq延遲消息

實(shí)現(xiàn)原理

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

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

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

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

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

優(yōu)缺點(diǎn)

優(yōu)點(diǎn)：

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

缺點(diǎn)：

• 引入額外的消息隊(duì)列，增加項(xiàng)目的維護(hù)和復(fù)雜度。
• 支持固定時(shí)長的消息延遲，針對(duì)任意時(shí)長的消息延遲需要進(jìn)行擴(kuò)展。

總結(jié)

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

最新評(píng)論