<?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>Remarque : Gestion des problèmes de compatibilité avec Redisson et Springboot# 🎜 🎜#
Fichier de configurationIl existe trois façons d'intégrer Redisson avec Springbootspring:
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);
}
}
}
Analyse des principes
RedissonDelayedQueue Nous voyons quatre caractères 
# 🎜🎜#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
#🎜. 🎜#redisson_delay_queue:xxx, type de données liste, je n'ai trouvé aucune utilité pour le moment, mais il sera écrit ici lors de la soumission de la tâche, et les éléments à l'intérieur seront supprimés lorsque la file d'attente sera transféré# 🎜🎜##🎜🎜##🎜🎜##🎜🎜#xxx : type de données de liste, appelé file d'attente cible. Les tâches stockées dans cette file d'attente ont atteint le délai et peuvent être consommées. . La tâche obtenue par l'utilisateur, donc la méthode take de RBlockingQueue dans la démo ci-dessus obtient la tâche de cette file d'attente cible #🎜🎜##🎜🎜##🎜🎜##🎜🎜#redisson_delay_queue_channel:xxx , est un canal utilisé pour notifier au client de démarrer une tâche retardée#🎜🎜##🎜🎜##🎜🎜##🎜🎜#Queue Creation#🎜🎜##🎜🎜#
RedissonDelayedQueue Delay file d'attente Lors de la création, le service de transfert de file d'attente et les quatre clés de correction des couleurs importantes pour la mise en œuvre de la file d'attente retardée sont spécifiés. Le code principal consiste à spécifier la tâche de transfert de file d'attente#🎜🎜# 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));
#🎜🎜##🎜🎜#Core codeRedissonDelayedQueue#offerAsync#🎜🎜#rrreee#🎜🎜#consumer#🎜 🎜 ##🎜🎜#Le moyen le plus simple pour les consommateurs est de simplement lire BLPOP dans la liste sans préfixe#🎜🎜#Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!
Logiciel de base de données couramment utilisé
Que sont les bases de données en mémoire ?
Lequel a une vitesse de lecture plus rapide, mongodb ou redis ?
Comment utiliser Redis comme serveur de cache
Comment Redis résout la cohérence des données
Comment MySQL et Redis assurent-ils la cohérence des doubles écritures ?
Quelles données le cache Redis stocke-t-il généralement ?
Quels sont les 8 types de données de Redis
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
