<?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>Nota: Tangani isu keserasian redisson dan springboot
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: 8package 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);
}
}
}
Daripada pelaksanaan RedissonDelayedQueue kita lihat terdapat empat peranan
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
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('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);
}
};
Kod terasRedissonDelayedQueue#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));Atas ialah kandungan terperinci Cara Redis melaksanakan baris gilir kelewatan. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!
Perisian pangkalan data yang biasa digunakan
Apakah pangkalan data dalam memori?
Yang manakah mempunyai kelajuan bacaan yang lebih pantas, mongodb atau redis?
Cara menggunakan redis sebagai pelayan cache
Bagaimana redis menyelesaikan ketekalan data
Bagaimanakah mysql dan redis memastikan konsistensi penulisan dua kali?
Apakah data yang biasanya disimpan oleh redis cache?
Apakah 8 jenis data redis
![[Web front-end] Permulaan pantas Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
