生產redisson延時隊列不消費問題排查解決

更新時間：2022年09月26日 15:57:07 作者：法力損毀

這篇文章主要為大家介紹了生產redisson延時隊列不消費問題排查解決，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

問題描述

項目使用redisson延時隊列功能，實現(xiàn)直播的開播提醒，突然有一天業(yè)務爆出問題，未觸發(fā)開播提醒。

初步排查

首先通過查詢生產日志，發(fā)送端日志存在，沒有消費日志，猜測消費端沒有消費到延時消息，，在dba的協(xié)助下查詢redis隊列，消息也確實存在，但已經過了過期時間，由此證明redisson消費者出現(xiàn)問題。通過服務日志發(fā)現(xiàn)在最后一次設置自定義推送任務是在一次服務發(fā)布之前，服務發(fā)布后，之前設置的自定義推送消息均沒有被客戶端消費，由此猜想是由發(fā)布服務導致消費端失效。

排查過程

發(fā)送端代碼

public <T> void produce(String delayQueue, T t, long delay, TimeUnit timeUnit) {
    try {
        log.info("delay msg,delayQueue:{},key:{},delay:{}", delayQueue, t, delay);
        if (delay < 0) {
            delay = 0;
        }
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(delayQueue);
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
    }catch (Exception e){
        log.error("添加延時任務隊列失敗",e);
    }
}

消費端代碼

public class DelayTaskHandler implements Runnable {
    @Override
    public void run() {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(delayQueue);
        while (true) {
            try {
                T value = blockingFairQueue.take();
                log.info("delay queue {},延時任務開始執(zhí)行,value - {} , timeStamp - {} , threadName - {}", delayQueue, value, System.currentTimeMillis(), Thread.currentThread().getName());
                consumer.accept(value);
            } catch (Exception e) {
                log.error("延時任務執(zhí)行失敗,", e);
            }
        }
    }
}

因為redisson 延時隊列是基于redis實現(xiàn)的，所以從redis執(zhí)行命令開始入手排查

1.打開redis監(jiān)控，啟動服務,發(fā)現(xiàn)redis首先執(zhí)行了blpop命令，阻塞等待{cl-live-admin:notice_delay_queue} 隊列消息

2.提交一個延時任務后，觀察redis命令

此時發(fā)現(xiàn)redis首先執(zhí)行了一個SUBSCRIBE命令，訂閱了一個隊列，然后執(zhí)行了一段lua腳本，主要包括以下命令：

zrangebyscore:獲取zset中score在0至當前時間戳范圍內的前一百條數(shù)據如果獲取到數(shù)據則循環(huán)執(zhí)行rpush，lrem，zrem命令
zrange:取zset中第一條數(shù)據
zadd:向zset中添加一條數(shù)據，score為時間戳
rpush：向list右邊push一條數(shù)據
publish：如果添加的消息在頂部，則發(fā)布一條訂閱消息

3.消費一條消息

同樣消費的時候也是提交了一條lua腳本，主要執(zhí)行了以下命令可以看到和發(fā)送端命令相似

zrangebyscore：獲取zset中score在0至當前時間戳范圍內的前一百條數(shù)據
rpush：向list右邊push一條數(shù)據
lrem：刪除一條數(shù)據
zrem：刪除zeset中的數(shù)據
zrange：獲取第一條數(shù)據
BLPOP：阻塞等待隊列消息

通過以上redis命令的執(zhí)行可以發(fā)現(xiàn)一個命令SUBCRIBE用于訂閱redis的一個隊列，而這個命令只在發(fā)送消息的時候執(zhí)行了，在消費的時候沒有執(zhí)行。從而驗證了當服務重啟后如果沒有新的消息發(fā)送，那么客戶端就不會發(fā)送SUBCRIBE命令，訂閱延時隊列，這就導致在服務重啟前發(fā)送的消息到時間后無法消費。

解決方案

在消費端啟動的時候添加一行代碼用于訂閱延時隊列

 //訂閱redis隊列
 redissonClient.getDelayedQueue(blockingFairQueue);

那么為什么沒有訂閱就消費不到消息了呢？帶著疑問繼續(xù)深入理解redisson的實現(xiàn)

redisson 延時隊列原理

首先回到消費端代碼

在我們沒有發(fā)送訂閱命令的時候，客戶端只是在阻塞等待一個指定隊列的消息，那么這個隊列的消息是誰放進去的呢？帶著疑問我們再看發(fā)送端代碼

直接進入 delayedQueue.offer()方法內部

可以看到發(fā)送端是提交了一個lua腳本主要執(zhí)行了zadd，rpush，publish命令，這里我們需要注意publish命令，在redis中pub/sub是對應的，當有publish的時候，那么subcribe端會收到該訂閱消息。

那么是誰收到了訂閱的消息，收到消息后又做了什么呢，回到redissonClient.getDelayedQueue(blockingFairQueue)代碼中

繼續(xù)進入 new RedissonDelayedQueue()

可以看到這里創(chuàng)建了一個QueueTransferTask，實現(xiàn)了pushTaskAsync()方法，具體內容是一個lua腳本，首先執(zhí)行zrangebyscore 獲取過期的前一百條數(shù)據，循環(huán)調用rpush，lrem，zrem，注意這里rpush的隊列為我們指定的延時隊列，也就是consumer端take的隊列。至此明白了消費端的消息是方法pushTaskAsync()執(zhí)行后放入的。那么什么時候執(zhí)行這個方法呢。

進入 queueTransferService.schedule(queueName, task)方法