Redis優雅地實現延遲佇列的方法分享

2023-02-27 06:01:06

前言

工作中常常會遇到這樣的場景，如訂單到期未支付取消，到期自動續費等，我們發現延遲佇列非常適合在這樣的場景中使用。常見的延遲佇列的優秀實現有rabbitMQ的死信佇列，RocketMQ的延遲佇列等，但是了有時候專案沒有特別的大，沒有引入類似的訊息中介軟體，但是了又遇到了特別適合使用延遲佇列的場景，我們一般會利用已有的redis實現一個簡陋的延遲佇列。常見的實現方式有監聽過期key，使用zset利用分值進行一個匹配，但是了這些實現或多或少有些問題，不夠優雅。監聽過期key是一種危險行為，一是如果過redis中key數量較大監聽過期key可能導致服務負載異常，二是redis中key過期後key是惰性刪除的，因此監聽機制需要主動觸發。利用zset分值實現呢，需要自己開發程式碼處理定時輪訓以及key刪除的邏輯，具有一定的工作量和複雜度。哪有沒有一種優雅的redis延遲佇列的實現呢？

Redisson是Redis伺服器上的分散式可伸縮Java資料結構----駐記憶體資料網格(In-Memory Data Grid，IMDG)。底層使用netty框架，並提供了與java物件相對應的分散式物件、分散式集合、分散式鎖和同步器、分散式服務等一系列的Redisson的分散式物件。為我們提供了許多開箱即用的功能。今天介紹Redisson實現的優雅的延遲佇列。

使用

依賴設定

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.12.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.homeey</groupId>
    <artifactId>redis-delay-queue</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>redis-delay-queue</name>
    <description>redis-delay-queue</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson-spring-boot-starter</artifactId>
            <version>3.19.3</version>
        </dependency>
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson-spring-data-23</artifactId>
            <version>3.19.3</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

備註：處理redisson和springboot相容性問題

組態檔

springboot整合redisson有三種方式

第一種：通用的redis設定+redisson的自動設定[最簡單]
第二種：使用單獨的redisson組態檔
第三種：使用spring.redis.redisson這個設定key下進行設定

詳細的整合檢視 springboot整合redisson設定

spring:
  redis:
    database: 0
    host: localhost
    port: 6379
    timeout: 10000
    lettuce:
      pool:
        max-active: 8
        max-wait: -1
        min-idle: 0
        max-idle: 8

demo程式碼

package com.homeey.redisdelayqueue.delay;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 明天的你會因今天到的努力而幸運
 *
 * @author jt4mrg@qq.com
 * 23:11 2023-02-19 2023
 **/
@Slf4j
@Component
@RequiredArgsConstructor
public class RedissonDelayQueue {

    private final RDelayedQueue<String> delayedQueue;
    private final RBlockingQueue<String> blockingQueue;


    @PostConstruct
    public void init() {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        executorService.submit(() -> {
            while (true) {
                try {
                    String task = blockingQueue.take();
                    log.info("rev delay task:{}", task);
                } catch (Exception e) {
                    log.error("occur error", e);
                }
            }
        });
    }

    public void offerTask(String task, long seconds) {
        log.info("add delay task:{},delay time:{}s", task, seconds);
        delayedQueue.offer(task, seconds, TimeUnit.SECONDS);
    }


    @Configuration
    static class RedissonDelayQueueConfigure {

        @Bean
        public RBlockingQueue<String> blockingQueue(RedissonClient redissonClient) {
            return redissonClient.getBlockingQueue("TOKEN-RENEWAL");
        }

        @Bean
        public RDelayedQueue<String> delayedQueue(RBlockingQueue<String> blockingQueue,
                                                  RedissonClient redissonClient) {
            return redissonClient.getDelayedQueue(blockingQueue);
        }
    }
}

執行效果

原理分析

從RedissonDelayedQueue實現中我們看到有四個角色

redisson_delay_queue_timeout:xxx，sorted set資料型別，存放所有延遲任務，按照延遲任務的到期時間戳（提交任務時的時間戳 + 延遲時間）來排序的，所以列表的最前面的第一個元素就是整個延遲佇列中最早要被執行的任務，這個概念很重要
redisson_delay_queue:xxx，list資料型別，暫時沒發現什麼用，只是在提交任務時會寫入這裡面，佇列轉移時又會刪除裡面的元素
xxx：list資料型別，被稱為目標佇列，這個裡面存放的任務都是已經到了延遲時間的，可以被消費者獲取的任務，所以上面demo中的RBlockingQueue的take方法是從這個目標佇列中獲取到任務的
redisson_delay_queue_channel:xxx，是一個channel，用來通知使用者端開啟一個延遲任務

佇列建立

RedissonDelayedQueue延遲佇列建立時，指定了佇列轉移服務，以及實現延遲佇列的四個重要校色的key。核心程式碼是指定佇列轉移任務

 QueueTransferTask task = new QueueTransferTask(commandExecutor.getConnectionManager()) {
            
            @Override
            protected RFuture<Long> pushTaskAsync() {
                return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
                        "local expiredValues = redis.call('zrangebyscore', KEYS[2], 0, ARGV[1], 'limit', 0, ARGV[2]); "//拿到zset中過期的值列表
                      + "if #expiredValues > 0 then " //如果有
                          + "for i, v in ipairs(expiredValues) do "
                              + "local randomId, value = struct.unpack('dLc0', v);"//解構訊息，在提交任務時打包的訊息
                              + "redis.call('rpush', KEYS[1], value);" //放入無字首的list 隊頭
                              + "redis.call('lrem', KEYS[3], 1, v);"//移除帶字首list 隊尾元素
                          + "end; "
                          + "redis.call('zrem', KEYS[2], unpack(expiredValues));" //移除zset中本次讀取的過期元素
                      + "end; "
                        // get startTime from scheduler queue head task
                      + "local v = redis.call('zrange', KEYS[2], 0, 0, 'WITHSCORES'); "//取zset最小分值的元素
                      + "if v[1] ~= nil then "
                         + "return v[2]; " //返回分值，即過期時間
                      + "end "
                      + "return nil;",
                      Arrays.asList(getRawName(), timeoutSetName, queueName),
                      System.currentTimeMillis(), 100);
            }
            
            @Override
            protected RTopic getTopic() {
                return RedissonTopic.createRaw(LongCodec.INSTANCE, commandExecutor, channelName);
            }
        };

生產者

核心程式碼RedissonDelayedQueue#offerAsync

 return commandExecutor.evalWriteNoRetryAsync(getRawName(), codec, RedisCommands.EVAL_VOID,
                "local value = struct.pack('dLc0', tonumber(ARGV[2]), string.len(ARGV[3]), ARGV[3]);" //打包訊息體：訊息id，訊息長度，訊息值
              + "redis.call('zadd', KEYS[2], ARGV[1], value);"//zset中加入訊息及其超時分值
              + "redis.call('rpush', KEYS[3], value);" //向帶字首的list中新增訊息
              // if new object added to queue head when publish its startTime 
              // to all scheduler workers 
              + "local v = redis.call('zrange', KEYS[2], 0, 0); "//取出zset中第一個元素
              + "if v[1] == value then " //如果最快過期的元素就是這次傳送的訊息
                 + "redis.call('publish', KEYS[4], ARGV[1]); " //channel中釋出一下超時時間
              + "end;",
              Arrays.asList(getRawName(), timeoutSetName, queueName, channelName),
              timeout, randomId, encode(e));

消費者

消費者最簡單，直接從不帶字首的list中BLPOP讀取就可以

整個流程

總結思考

Lua是redis的好朋友，我們可以看到Redisson實現延遲佇列時，大量使用到lua指令碼，因Redis會將整個指令碼作為一個整體執行，中間不會被其他請求插入。因此在指令碼執行過程中無需擔心會出現競態條件，無需使用事務。我們在平時開發時有多個redis命令操作的有簡單的業務邏輯，不妨嘗試一下lua指令碼的方式，可以避免使用分散式鎖來保障一致性。

Redisson的原始碼值得一讀，有很多新東西值得學習，如果其用到的netty基於時間輪演演算法的定時任務排程，可以讓我們基於此實現自己的任務排程框架，也讓我有了去探究這種實現方式和基於ScheduledThreadPoolExecutor的定時排程的差異及各自優劣的慾望。

以上就是Redis優雅地實現延遲佇列的方法分享的詳細內容，更多關於Redis延遲佇列的資料請關注it145.com其它相關文章！