應(yīng)用場(chǎng)景

通常在一些需要經(jīng)歷一段時(shí)間或者到達(dá)某個(gè)指定時(shí)間節(jié)點(diǎn)才會(huì)執(zhí)行的功能，比如以下這些場(chǎng)景：

• 訂單超時(shí)提醒
• 收貨自動(dòng)確認(rèn)
• 會(huì)議提醒
• 代辦事項(xiàng)提醒

為什么使用延時(shí)隊(duì)列

對(duì)于數(shù)據(jù)量小且實(shí)時(shí)性要求不高的需求來說，最簡(jiǎn)單的方法就是定時(shí)掃描數(shù)據(jù)庫(kù)。

但是，當(dāng)數(shù)量達(dá)到數(shù)百萬、上千萬級(jí)別且時(shí)，定時(shí)掃庫(kù)就顯得非常低效且消耗資源，

甚至有些時(shí)間間隔小實(shí)時(shí)性要求高的情況，上一次掃描還沒結(jié)束，下一次就又開始了，

這時(shí)候如果使用延時(shí)隊(duì)列就會(huì)比較合適

延時(shí)隊(duì)列的幾種方式：

• Quartz 定時(shí)任務(wù)實(shí)現(xiàn)掃庫(kù)
• DelayQueue JDK中提供了一組實(shí)現(xiàn)延遲隊(duì)列的API
• Redis sorted set
• Redis 過期鍵監(jiān)聽回調(diào)
• RabbitMQ 死信隊(duì)列
• RabbitMQ 基于插件實(shí)現(xiàn)延遲隊(duì)列
• Wheel 時(shí)間輪訓(xùn)算法

Redisson 實(shí)現(xiàn)延時(shí)隊(duì)列

顧名思義 Redis son 就是 Redis 的兒子，舉個(gè)栗子先：

1.引入 pom

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>${lastest.version}</version>
</dependency>

2.封裝一個(gè) RedissonQueue 類

@Service
public class RedissonQueue {

    public static final String QUEUE = "delayQueue";

    // 默認(rèn)超時(shí)時(shí)間，30秒
    public static final Integer DEFAULT_TIMEOUT = 30;

    @Resource
    private RedissonClient redissonClient;

    // 加入任務(wù)并設(shè)置到期時(shí)間
    public void offer(String taskId, Integer timeout) {
        RDelayedQueue<String> delayedQueue = delayedQueue();
        delayedQueue.offer(taskId, Objects.isNull(timeout) ? DEFAULT_TIMEOUT : timeout, TimeUnit.SECONDS);
    }

    // 移除任務(wù)
    public void remove(String taskId) {
        RDelayedQueue<String> delayedQueue = delayedQueue();
        delayedQueue.removeIf(messageId -> messageId.equals(taskId));
    }

    // 任務(wù)列表
    public RDelayedQueue<String> delayedQueue() {
        RBlockingDeque<String> blockingDeque = blockingDeque();
        return redissonClient.getDelayedQueue(blockingDeque);
    }

    public RBlockingDeque<String> blockingDeque() {
        return redissonClient.getBlockingDeque(QUEUE);
    }

    public boolean isShutdown() {
        return redissonClient.isShutdown();
    }

    public void shutdown() {
        redissonClient.shutdown();
    }

}

3.交給 Spring 管理

@Slf4j
@Service
public class RedissonService implements ApplicationRunner {

    @Resource
    private RedissonQueue redissonQueue;

    @Resource(name = "threadPoolTaskExecutor")
    private ThreadPoolTaskExecutor executor;

    @Override
    public void run(ApplicationArguments args) {
        RBlockingDeque<String> blockingDeque = redissonQueue.blockingDeque();
        executor.execute(() -> {
            while (true) {
                if (redissonQueue.isShutdown()) {
                    return;
                } else {
                    String messageId = null;
                    try {
                        messageId = blockingDeque.take();
                    } catch (InterruptedException e) {
                        log.warn("RedissonConsumer error:{}", e.getMessage());
                    }
                    if (!Objects.isNull(messageId) && !messageId.isEmpty()) {
                        log.warn("timeout messageId : {}", messageId);
                    }
                }
            }
        });

    }

    // 初始化，啟動(dòng)服務(wù)就執(zhí)行一次
    @PostConstruct
    public void init() {
        redissonQueue.delayedQueue();
    }

    @PreDestroy
    public void shutdown() {
        redissonQueue.shutdown();
    }

}

4.測(cè)試接口

@Operation(summary = "添加任務(wù)", description = "添加任務(wù)")
@PostMapping
public ResponseEntity<?> add(@RequestParam(value = "taskId", required = false) String taskId,
                             @RequestParam(value = "timeout", required = false) Integer timeout) {
    taskId = StringUtils.isEmpty(taskId) ? String.valueOf(snowflake.nextId()) : taskId;
    redissonQueue.offer(taskId, timeout);
    return ResponseEntity.ok().body(redissonQueue.delayedQueue());
}

@Operation(summary = "移除任務(wù)", description = "移除任務(wù)")
@DeleteMapping(value = "/{taskId}")
public ResponseEntity<?> remove(@PathVariable("taskId") String taskId) {
    redissonQueue.remove(taskId);
    return ResponseEntity.ok().body(redissonQueue.delayedQueue());
}

5.測(cè)試結(jié)果

添加10個(gè)任務(wù)

刪除第1個(gè)任務(wù)

可以看到第一個(gè)任務(wù)刪除后沒有被執(zhí)行（沒有設(shè)置到期時(shí)間，默認(rèn)為30秒到期）

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

• redisson_delay_queue_timeout:delayQueue，sorted set 數(shù)據(jù)類型，存放所有延遲任務(wù)，按延遲任務(wù)的到期時(shí)間戳（提交任務(wù)時(shí)間戳 +
  延遲時(shí)間）排序，所以列表最前面第一個(gè)元素就是整個(gè)延遲隊(duì)列中最早被執(zhí)行的任務(wù)。
• redisson_delay_queue:delayQueue，list 數(shù)據(jù)類型，也是存放所有任務(wù)。
• delayQueue，list 數(shù)據(jù)類型，被稱為目標(biāo)隊(duì)列，這個(gè)里面存放的任務(wù)都是已經(jīng)到延遲時(shí)間的，可以被消費(fèi)者獲取的任務(wù)，所以上面示例中
  RBlockingQueue 的 take 方法是從此目標(biāo)隊(duì)列中獲取任務(wù)的。
• redisson_delay_queue_channel:delayQueue，是一個(gè) channel，用來通知客戶端開啟一個(gè)延遲任務(wù)
• 生產(chǎn)者提交任務(wù)時(shí)將任務(wù)放到 redisson_delay_queue_timeout:delayQueue 中，提交任務(wù)的時(shí)間戳+延遲時(shí)間
• 客戶端會(huì)有一個(gè)延遲任務(wù)，這個(gè)延遲任務(wù)會(huì)向 Redis Server 發(fā)送一段 lua 腳本，Redis 執(zhí)行 lua 腳本中的命令，此操作是原子性的

lua 腳本主要干兩件事

• 將到了延遲時(shí)間的任務(wù)從 redisson_delay_queue_timeout:delayQueue 中移除，存到 delayQueue 這個(gè)目標(biāo)隊(duì)列
• 獲取到 redisson_delay_queue_timeout:delayQueue 中最早到期時(shí)間的任務(wù)的到期時(shí)間戳，發(fā)布到 redisson_delay_queue_channel:
  delayQueue channel 中

當(dāng)客戶端監(jiān)聽到 redisson_delay_queue_channel:delayQueue 這個(gè) channel 的消息時(shí)，會(huì)再次提交一個(gè)客戶端延遲任務(wù)，延遲時(shí)間就是消息（最早到期時(shí)間任務(wù)的到期時(shí)間戳）當(dāng)前時(shí)間戳
這個(gè)時(shí)間其實(shí)也就是 redisson_delay_queue_channel:delayQueue 中最早到期時(shí)間的任務(wù)的剩余的延遲時(shí)間。
一旦時(shí)間來到最早到期時(shí)間任務(wù)的到期時(shí)間戳，redisson_delay_queue_timeout:delayQueue 中最早到期時(shí)間的任務(wù)已經(jīng)到期，客戶端的延遲任務(wù)也同時(shí)到期，
于是開始執(zhí)行 lua 腳本操作，及時(shí)將到期任務(wù)放到目標(biāo)隊(duì)列中。然后再次發(fā)布剩余的延遲任務(wù)中最早到期任務(wù)的到期時(shí)間戳到 channel
中，
如此循環(huán)運(yùn)行下去，保證 redisson_delay_queue_timeout:delayQueue 中到期數(shù)據(jù)能及時(shí)放到目標(biāo)隊(duì)列中。
這里存在一個(gè)特殊情況，需要項(xiàng)目啟動(dòng)時(shí)就執(zhí)行一次延時(shí)隊(duì)列。因?yàn)橛捎跊]有客戶端延遲任務(wù)的執(zhí)行，
可能會(huì)出現(xiàn) redisson_delay_queue_timeout:delayQueue 隊(duì)列中有到期但是沒有被放到目標(biāo)隊(duì)列的可能，啟動(dòng)就執(zhí)行一次是為了保證到期的數(shù)據(jù)能被及時(shí)放到目標(biāo)隊(duì)列中。

結(jié)論

• Redisson 方案理論上沒有延遲，但當(dāng)消息數(shù)量劇增，消費(fèi)者消費(fèi)緩慢這種情況下，可能會(huì)導(dǎo)致延遲任務(wù)消費(fèi)的延遲。

• 消息丟失問題 Redisson 方案最大程度上減輕消息丟失的可能性，因?yàn)樗腥蝿?wù)都是存在 list 和 sorted set 兩種數(shù)據(jù)類型中，Redis
  有持久化機(jī)制。除非整個(gè) redis 集群宕機(jī)，可能丟失一小部分?jǐn)?shù)據(jù)。

• 廣播任務(wù)問題，是不會(huì)出現(xiàn)的，因?yàn)槊總€(gè)客戶端都是從同一個(gè)目標(biāo)隊(duì)列中獲取任務(wù)。

Redisson 這種實(shí)現(xiàn)方案是比較合適且靠譜的，一般中小型項(xiàng)目建議用 Redisson 實(shí)現(xiàn)延遲隊(duì)列，規(guī)模較大的項(xiàng)目直接上 MQ。

整合DEMO倉(cāng)庫(kù)地址

到此這篇關(guān)于springboot整合redisson實(shí)現(xiàn)延時(shí)隊(duì)列(附倉(cāng)庫(kù)地址)的文章就介紹到這了,更多相關(guān)springboot redisson延時(shí)隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論