ホームページ > データベース > Redis > Redis が遅延キューを実装する方法

Redis が遅延キューを実装する方法

WBOY
リリース: 2023-05-26 20:39:28
転載
1584 人が閲覧しました

    使用

    依存関係構成

    <?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 を統合するには 3 つの方法があります

    • 最初の: 一般的な redis 構成 redisson 自動構成 [最も単純な]

    • 2 番目のタイプ: 別の redisson 構成ファイルを使用します

    • 3 番目のタイプ: spring.redis.redisson を使用して構成キーの下で構成します

    詳細な統合、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
    ログイン後にコピー

    デモ コード

    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実装では 4 つのロールが表示されます

    Redis が遅延キューを実装する方法

    • #redisson_lay_queue_timeout:xxx、ソート セット データ タイプ、ストアすべての遅延タスク。遅延タスクの有効期限タイムスタンプ (タスクが送信されたときのタイムスタンプ遅延時間) に従って並べ替えられるため、リストの先頭の最初の要素が遅延キュー全体で実行される最も早いタスクになります。概念は非常に重要です

    • redisson_delay_queue:xxx、リストデータ型、今のところ使い道が見つかりませんが、提出時にここに書きます。タスク、キュー転送 内部の要素は削除されます

    • xxx: 対象キューと呼ばれるリスト データ型 このキューに格納されているタスクはすべて到達しましたコンシューマーが取得できるタスクであるため、上記のデモの RBlockingQueue の take メソッドは、このターゲット キューからタスク

    • redisson_lay_queue_channel:xxx を取得します。これは、遅延タスクを開始するようにクライアントに通知するために使用されるチャネルです。

    キューの作成

    RedissonDelayedQueue遅延キューが作成されると、キュー転送サービスと、遅延キューを実装するための 4 つの重要な色補正キーが指定されています。コア コードは、キュー転送タスク

     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);
                }
            };
    ログイン後にコピー

    Producer

    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));
    ログイン後にコピー
    ##を指定することです。 #Consumer

    コンシューマーにとって最も簡単な方法は、プレフィックスなしでリストから BLPOP を読み取ることです

    以上がRedis が遅延キューを実装する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

    関連ラベル:
    ソース:yisu.com
    このウェブサイトの声明
    この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。
    人気のチュートリアル
    詳細>
    最新のダウンロード
    詳細>
    ウェブエフェクト
    公式サイト
    サイト素材
    フロントエンドテンプレート