延時隊列-基于Redisson實現(xiàn)的延時隊列

前言

定時調(diào)度基本是每個項目都會遇到的業(yè)務(wù)場景，一般地，都會通過任務(wù)調(diào)度工具執(zhí)行定時任務(wù)完成，定時任務(wù)有兩點缺陷，一、定時任務(wù)執(zhí)行頻度限制，實際執(zhí)行的時間可能會晚于理想的設(shè)定時間，例如，如果要通過定時任務(wù)實現(xiàn)在下單后15分鐘仍未支付則取消訂單的功能，假設(shè)定時任務(wù)的執(zhí)行頻度為每分鐘執(zhí)行一次，對于有些訂單而言，其實際取消時間是介于15-16分鐘之間，不夠精確；二、定時任務(wù)執(zhí)行需要時間，定時任務(wù)的執(zhí)行也需要時間，如果業(yè)務(wù)場景的數(shù)據(jù)量較大，執(zhí)行一次定時任務(wù)需要足夠長的時間，進一步放大了缺點一。

Redisson延時隊列

在上一篇文章中延時隊列-基于RabbitMq延時消息插件實現(xiàn)的延時隊列
介紹了一種基于RabbitMq延時消息插件實現(xiàn)的延時隊列，除此之外，另一種比較常見的實現(xiàn)方式是Redisson延時隊列。
在Redisson官網(wǎng)文檔中，使用延時隊列的代碼如下：

RQueue<String> distinationQueue = ...
RDelayedQueue<String> delayedQueue = getDelayedQueue(distinationQueue);
// 10秒鐘以后將消息發(fā)送到指定隊列
delayedQueue.offer("msg1", 10, TimeUnit.SECONDS);
// 一分鐘以后將消息發(fā)送到指定隊列
delayedQueue.offer("msg2", 1, TimeUnit.MINUTES);

當(dāng)該延時隊列不需要使用時，應(yīng)當(dāng)主動銷毀

RDelayedQueue<String> delayedQueue = ...
delayedQueue.destroy();

Redisson的延時隊列是對另一個隊列的再包裝，使用時要先將延時消息添加到延時隊列中，當(dāng)延時隊列中的消息達到設(shè)定的延時時間后，該延時消息才會進行進入到被包裝隊列中，因此，我們只需要對被包裝隊列進行監(jiān)聽即可。

代碼部分

在上一篇文章中，我們定義了DelayMessageManager延時消息管理器并提供了一種基于RabbitMq的實現(xiàn)，本文中我們再提供一種基于Redisson延時隊列的實現(xiàn)即可。

package com.cube.share.delay.manager;

import cn.hutool.core.thread.ThreadUtil;
import com.cube.share.delay.handler.DelayMessageHandler;
import com.cube.share.delay.message.DelayMessage;
import com.cube.share.delay.message.DelayMessageType;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static com.cube.share.base.utils.SpringContextUtil.getBean;

/**
 * @author cube.li
 * @date 2021/10/11 20:59
 * @description 基于redisson實現(xiàn)的延時消息管理器
 */
@Component
@Slf4j
public class RedissonDelayMessageManager implements DelayMessageManager {

    private RedissonClient redissonClient = Redisson.create();

    private RBlockingQueue<DelayMessage> rBlockingQueue = redissonClient.getBlockingDeque("redisson-delay-message-queue");

    private RDelayedQueue<DelayMessage> rDelayedQueue = redissonClient.getDelayedQueue(rBlockingQueue);

    private final Map<DelayMessageType, DelayMessageHandler> handlerMap = new ConcurrentHashMap<>(16);

    @Override
    public void add(DelayMessage message) {
        if (rDelayedQueue.contains(message)) {
            return;
        }
        log.info("redisson-delay-message-queue,add message = {}", message);
        rDelayedQueue.offer(message, message.getProperties().getExpire(), message.getProperties().getTimeUnit());
    }

    @Override
    public boolean remove(DelayMessage message) {
        return rDelayedQueue.remove(message);
    }

    @Override
    public void destroy() {
        rDelayedQueue.destroy();
    }

    @Override
    public void afterPropertiesSet() {
        Arrays.stream(DelayMessageType.values()).forEach(delayMessageType -> handlerMap.put(delayMessageType, getBean(delayMessageType.getHandler())));
        Thread thread = new Thread(() -> {
            while (true) {
                try {
                    DelayMessage delayMessage = rBlockingQueue.take();
                    log.info("redisson-delay-message-queue,consume message = {}", delayMessage);
                    handlerMap.get(delayMessage.getType()).handle(delayMessage);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    ThreadUtil.safeSleep(1000);
                    afterPropertiesSet();
                }
            }
        });
        thread.setDaemon(true);
        thread.start();
    }
}

代碼中，首先創(chuàng)建了一個Redisson實現(xiàn)的阻塞隊列RBlockingQueue的實例rBlockingQueue，然后又使用該阻塞隊列rBlockingQueue創(chuàng)建了一個延時隊列RDelayedQueue的實例rDelayedQueue，按照前文中的描述，延時消息添加后并不是立即進入到阻塞隊列rBlockingQueue中，而是到達了設(shè)定的延時時間之后才會從延時隊列rDelayedQueue進入到阻塞隊列rBlockingQueue；因此，延時消息的添加由延時隊列rDelayedQueue完成而延時隊列的消費則由阻塞隊列rBlockingQueue完成，注意，這里如果直接對延時隊列rDelayedQueue進行監(jiān)聽，則延時消息剛加入時就會被消費，達不到延時的效果。
相比于Redisson官網(wǎng)文檔延時隊列中給出的代碼示例，這里被包裝隊列使用阻塞隊列RBlockingQueue的好處是rBlockingQueue.take()會一直阻塞直至隊列內(nèi)有可消費延時消息，避免無意義的循環(huán)占用CPU。

測試

單元測試代碼與上一篇文章基本相同，只是將DelayMessageManager的實現(xiàn)換成了RedissonDelayMessageManager

package com.cube.share.delay.manager;

import com.cube.share.base.utils.JacksonUtils;
import com.cube.share.delay.entity.ExecuteTask;
import com.cube.share.delay.entity.QrCode;
import com.cube.share.delay.message.DelayMessage;
import com.cube.share.delay.message.DelayMessageType;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author cube.li
 * @date 2021/9/22 17:12
 * @description 測試
 */
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class DelayMessageManagerTest {

    @Resource(type = RedissonDelayMessageManager.class)
    DelayMessageManager delayMessageManager;

    @Test
    void add() throws JsonProcessingException {
        DelayMessage delayMessage = new DelayMessage();
        QrCode qrCode = new QrCode();
        qrCode.setConfigId("fadfdaf110");
        qrCode.setUrl("http://www.baidu.com");
        delayMessage.setBody(JacksonUtils.toJsonString(qrCode));
        delayMessage.setType(DelayMessageType.DELETE_QR_CODE);
        DelayMessage.DelayMessageProperties properties = new DelayMessage.DelayMessageProperties();
        properties.setExpire(10);
        properties.setTimeUnit(TimeUnit.SECONDS);
        delayMessage.setProperties(properties);
        delayMessageManager.add(delayMessage);


        DelayMessage delayMessage1 = new DelayMessage();
        QrCode qrCode1 = new QrCode();
        qrCode1.setConfigId("fadfdaf1405");
        qrCode1.setUrl("http://www.baidu.com");
        delayMessage1.setBody(JacksonUtils.toJsonString(qrCode1));
        delayMessage1.setType(DelayMessageType.DELETE_QR_CODE);
        DelayMessage.DelayMessageProperties properties1 = new DelayMessage.DelayMessageProperties();
        properties1.setExpire(5);
        properties1.setTimeUnit(TimeUnit.SECONDS);
        delayMessage1.setProperties(properties1);
        delayMessageManager.add(delayMessage1);

        DelayMessage delayMessage2 = new DelayMessage();
        ExecuteTask task = new ExecuteTask();
        task.setId(1L);
        delayMessage2.setBody(JacksonUtils.toJsonString(task));
        delayMessage2.setType(DelayMessageType.EXECUTE_TASK);

        DelayMessage.DelayMessageProperties properties2 = new DelayMessage.DelayMessageProperties();
        properties2.setExpire(9);
        properties2.setTimeUnit(TimeUnit.SECONDS);
        delayMessage2.setProperties(properties2);
        delayMessage2.setSerialId(11235813L);
        delayMessageManager.add(delayMessage2);
        //delayMessageManager.remove(delayMessage2);

    }
}

啟動項目，并執(zhí)行單元測試
向延時隊列中加入三條延時分別為5s 9s 10s 的延時消息，單元測試控制臺輸出如下：

2021-10-12 22:26:40.690  INFO 21340 --- [           main] c.c.s.d.m.RedissonDelayMessageManager    : redisson-delay-message-queue,add message = DelayMessage(body={"url":"http://www.baidu.com","configId":"fadfdaf110"}, type=DELETE_QR_CODE, serialId=null, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=10))
2021-10-12 22:26:40.693  INFO 21340 --- [           main] c.c.s.d.m.RedissonDelayMessageManager    : redisson-delay-message-queue,add message = DelayMessage(body={"url":"http://www.baidu.com","configId":"fadfdaf1405"}, type=DELETE_QR_CODE, serialId=null, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=5))
2021-10-12 22:26:40.698  INFO 21340 --- [           main] c.c.s.d.m.RedissonDelayMessageManager    : redisson-delay-message-queue,add message = DelayMessage(body={"id":1}, type=EXECUTE_TASK, serialId=11235813, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=9))

項目的控制臺輸出如下：

2021-10-12 22:26:45.719  INFO 10452 --- [      Thread-11] c.c.s.d.m.RedissonDelayMessageManager    : redisson-delay-message-queue,consume message = DelayMessage(body={"url":"http://www.baidu.com","configId":"fadfdaf1405"}, type=DELETE_QR_CODE, serialId=null, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=5))
2021-10-12 22:26:45.719  INFO 10452 --- [      Thread-11] c.c.s.d.h.QrCodeDelayMessageHandler      : 二維碼延時消息處理中,message = DelayMessage(body={"url":"http://www.baidu.com","configId":"fadfdaf1405"}, type=DELETE_QR_CODE, serialId=null, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=5))
2021-10-12 22:26:49.719  INFO 10452 --- [      Thread-11] c.c.s.d.m.RedissonDelayMessageManager    : redisson-delay-message-queue,consume message = DelayMessage(body={"id":1}, type=EXECUTE_TASK, serialId=11235813, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=9))
2021-10-12 22:26:49.719  INFO 10452 --- [      Thread-11] c.c.s.d.h.ExecuteTaskDelayMessageHandler : 任務(wù)延時消息處理中,message=DelayMessage(body={"id":1}, type=EXECUTE_TASK, serialId=11235813, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=9))
2021-10-12 22:26:50.718  INFO 10452 --- [      Thread-11] c.c.s.d.m.RedissonDelayMessageManager    : redisson-delay-message-queue,consume message = DelayMessage(body={"url":"http://www.baidu.com","configId":"fadfdaf110"}, type=DELETE_QR_CODE, serialId=null, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=10))
2021-10-12 22:26:50.719  INFO 10452 --- [      Thread-11] c.c.s.d.h.QrCodeDelayMessageHandler      : 二維碼延時消息處理中,message = DelayMessage(body={"url":"http://www.baidu.com","configId":"fadfdaf110"}, type=DELETE_QR_CODE, serialId=null, properties=DelayMessage.DelayMessageProperties(timeUnit=SECONDS, expire=10))

在經(jīng)過5s 9s 10s后，對應(yīng)的消息均成功被消費。

總結(jié)

在上一篇基于RabbitMq延時消息插件實現(xiàn)的延時隊列之后，本文又提供了一種基于Redisson實現(xiàn)的延時隊列，大體上能夠滿足各種業(yè)務(wù)場景的需求；希望大家在寫業(yè)務(wù)代碼時能夠舉一反三，不止于代碼，不止于業(yè)務(wù)，才能在程序員的職業(yè)生涯里有更進一步的發(fā)展。
示例代碼

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容