首頁 > 資料庫 > Redis > 主體

Redis如何實現延遲隊列

WBOY
發布: 2023-05-26 20:39:28
轉載
1552 人瀏覽過

    使用

    依賴設定

    <?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);
            }
        }
    }
    登入後複製

    執行效果

    Redis如何實現延遲隊列

    原理分析

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

    Redis如何實現延遲隊列

    • 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(&#39;zrangebyscore&#39;, KEYS[2], 0, ARGV[1], &#39;limit&#39;, 0, ARGV[2]); "//拿到zset中过期的值列表
                          + "if #expiredValues > 0 then " //如果有
                              + "for i, v in ipairs(expiredValues) do "
                                  + "local randomId, value = struct.unpack(&#39;dLc0&#39;, v);"//解构消息,在提交任务时打包的消息
                                  + "redis.call(&#39;rpush&#39;, KEYS[1], value);" //放入无前缀的list 队头
                                  + "redis.call(&#39;lrem&#39;, KEYS[3], 1, v);"//移除带前缀list 队尾元素
                              + "end; "
                              + "redis.call(&#39;zrem&#39;, KEYS[2], unpack(expiredValues));" //移除zset中本次读取的过期元素
                          + "end; "
                            // get startTime from scheduler queue head task
                          + "local v = redis.call(&#39;zrange&#39;, KEYS[2], 0, 0, &#39;WITHSCORES&#39;); "//取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);
                }
            };
    登入後複製

    生產者

    Redis如何實現延遲隊列

    #核心程式碼RedissonDelayedQueue#offerAsync

    #
     return commandExecutor.evalWriteNoRetryAsync(getRawName(), codec, RedisCommands.EVAL_VOID,
                    "local value = struct.pack(&#39;dLc0&#39;, tonumber(ARGV[2]), string.len(ARGV[3]), ARGV[3]);" //打包消息体:消息id,消息长度,消息值
                  + "redis.call(&#39;zadd&#39;, KEYS[2], ARGV[1], value);"//zset中加入消息及其超时分值
                  + "redis.call(&#39;rpush&#39;, KEYS[3], value);" //向带前缀的list中添加消息
                  // if new object added to queue head when publish its startTime 
                  // to all scheduler workers 
                  + "local v = redis.call(&#39;zrange&#39;, KEYS[2], 0, 0); "//取出zset中第一个元素
                  + "if v[1] == value then " //如果最快过期的元素就是这次发送的消息
                     + "redis.call(&#39;publish&#39;, KEYS[4], ARGV[1]); " //channel中发布一下超时时间
                  + "end;",
                  Arrays.asList(getRawName(), timeoutSetName, queueName, channelName),
                  timeout, randomId, encode(e));
    登入後複製

    消費者

    消費者最簡單,直接從不帶前綴的list中BLPOP讀取就可以

    以上是Redis如何實現延遲隊列的詳細內容。更多資訊請關注PHP中文網其他相關文章!

    相關標籤:
    來源:yisu.com
    本網站聲明
    本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn
    熱門教學
    更多>
    最新下載
    更多>
    網站特效
    網站源碼
    網站素材
    前端模板