Cara Redis melaksanakan baris gilir kelewatan-Redis-php.cn

Gunakan

Konfigurasi kebergantungan

<?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>

Salin selepas log masuk

Nota: Tangani isu keserasian redisson dan springboot

Fail konfigurasi

Terdapat tiga cara untuk mengintegrasikan redisson dengan springboot

Yang pertama: konfigurasi redis umum + konfigurasi automatik redisson [paling mudah]
Kedua: Gunakan fail konfigurasi redisson yang berasingan
Ketiga: Gunakan spring.redis.redisson untuk mengkonfigurasi di bawah kekunci konfigurasi

Paparan integrasi terperinci springboot konfigurasi redisson bersepadu

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

Salin selepas log masuk

kod 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);
        }
    }
}

Salin selepas log masuk

kesan pelaksanaan

Cara Redis melaksanakan baris gilir kelewatan

analisis prinsip

Daripada pelaksanaan RedissonDelayedQueue kita lihat terdapat empat peranan

Cara Redis melaksanakan baris gilir kelewatan

redisson_delay_queue_timeout:xxx, jenis data set diisih, penyimpanan Semua tugas tertunda diisih mengikut kepada cap masa tamat tugas yang tertunda (cap masa semasa menyerahkan tugasan + masa tunda), jadi elemen pertama di bahagian hadapan senarai ialah tugasan paling awal untuk dilaksanakan dalam keseluruhan baris gilir tertunda Konsep ini sangat penting
redisson_delay_queue:xxx, senarai jenis data, saya tidak menemui apa-apa kegunaan buat masa ini, tetapi ia akan ditulis di sini semasa menghantar tugasan, dan apabila baris gilir dipindahkan Elemen di dalam akan dipadamkan
xxx: senarai jenis data, dipanggil baris gilir sasaran Tugasan yang disimpan dalam baris gilir ini telah mencapai masa tunda dan boleh menjadi Tugasan yang diperolehi oleh pengguna, jadi kaedah ambil RBlockingQueue dalam demo di atas memperoleh tugas daripada baris gilir sasaran ini
redisson_delay_queue_channel:xxx, iaitu saluran , digunakan untuk memberitahu pelanggan untuk memulakan tugas tertunda

Penciptaan baris gilir

RedissonDelayedQueueApabila baris gilir tertunda dibuat, perkhidmatan pemindahan baris gilir ditentukan dan empat langkah untuk melaksanakan baris gilir tertangguh ditentukan Kunci penting untuk pembetulan warna. Kod teras adalah untuk menentukan tugas pemindahan baris gilir

 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);
            }
        };

Salin selepas log masuk

Pengeluar

Cara Redis melaksanakan baris gilir kelewatan

Kod terasRedissonDelayedQueue#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));

Salin selepas log masuk

Pengguna

Perkara paling mudah untuk pengguna ialah hanya BLPOP daripada senarai tanpa awalan

Atas ialah kandungan terperinci Cara Redis melaksanakan baris gilir kelewatan. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!