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

2022-09-28 14:00:24

問題描述

專案使用redisson延時佇列功能，實現直播的開播提醒，突然有一天業務爆出問題，未觸發開播提醒。

初步排查

首先通過查詢生產紀錄檔，傳送端紀錄檔存在，沒有消費紀錄檔，猜測消費端沒有消費到延時訊息，，在dba的協助下查詢redis佇列，訊息也確實存在，但已經過了過期時間，由此證明redisson消費者出現問題。通過服務紀錄檔發現在最後一次設定自定義推播任務是在一次服務釋出之前，服務釋出後，之前設定的自定義推播訊息均沒有被使用者端消費，由此猜想是由釋出服務導致消費端失效。

排查過程

傳送端程式碼

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 {},延時任務開始執行,value - {} , timeStamp - {} , threadName - {}", delayQueue, value, System.currentTimeMillis(), Thread.currentThread().getName());
                consumer.accept(value);
            } catch (Exception e) {
                log.error("延時任務執行失敗,", e);
            }
        }
    }
}

因為redisson 延時佇列是基於redis實現的，所以從redis執行命令開始入手排查

1.開啟redis監控，啟動服務,發現redis首先執行了blpop命令，阻塞等待{cl-live-admin:notice_delay_queue} 佇列訊息

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

此時發現redis首先執行了一個SUBSCRIBE命令，訂閱了一個佇列，然後執行了一段lua指令碼，主要包括以下命令：

zrangebyscore:獲取zset中score在0至當前時間戳範圍內的前一百條資料如果獲取到資料則迴圈執行rpush，lrem，zrem命令
zrange:取zset中第一條資料
zadd:向zset中新增一條資料，score為時間戳
rpush：向list右邊push一條資料
publish：如果新增的訊息在頂部，則釋出一條訂閱訊息

3.消費一條訊息

同樣消費的時候也是提交了一條lua指令碼，主要執行了以下命令可以看到和傳送端命令相似

zrangebyscore：獲取zset中score在0至當前時間戳範圍內的前一百條資料
rpush：向list右邊push一條資料
lrem：刪除一條資料
zrem：刪除zeset中的資料
zrange：獲取第一條資料
BLPOP：阻塞等待佇列訊息

通過以上redis命令的執行可以發現一個命令SUBCRIBE用於訂閱redis的一個佇列，而這個命令只在傳送訊息的時候執行了，在消費的時候沒有執行。從而驗證了當服務重啟後如果沒有新的訊息傳送，那麼使用者端就不會傳送SUBCRIBE命令，訂閱延時佇列，這就導致在服務重啟前傳送的訊息到時間後無法消費。

解決方案

在消費端啟動的時候新增一行程式碼用於訂閱延時佇列

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

那麼為什麼沒有訂閱就消費不到訊息了呢？帶著疑問繼續深入理解redisson的實現

redisson 延時佇列原理

首先回到消費端程式碼

在我們沒有傳送訂閱命令的時候，使用者端只是在阻塞等待一個指定佇列的訊息，那麼這個佇列的訊息是誰放進去的呢？帶著疑問我們再看傳送端程式碼

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

可以看到傳送端是提交了一個lua指令碼主要執行了zadd，rpush，publish命令，這裡我們需要注意publish命令，在redis中pub/sub是對應的，當有publish的時候，那麼subcribe端會收到該訂閱訊息。

那麼是誰收到了訂閱的訊息，收到訊息後又做了什麼呢，回到redissonClient.getDelayedQueue(blockingFairQueue)程式碼中

繼續進入 new RedissonDelayedQueue()

可以看到這裡建立了一個QueueTransferTask，實現了pushTaskAsync()方法，具體內容是一個lua指令碼，首先執行zrangebyscore 獲取過期的前一百條資料，迴圈呼叫rpush，lrem，zrem，注意這裡rpush的佇列為我們指定的延時佇列，也就是consumer端take的佇列。至此明白了消費端的訊息是方法pushTaskAsync()執行後放入的。那麼什麼時候執行這個方法呢。

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