溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹“Redis如何實現延遲隊列”的相關知識，小編通過實際案例向大家展示操作過程，操作方法簡單快捷，實用性強，希望這篇“Redis如何實現延遲隊列”文章能幫助大家解決問題。
<?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
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);
}
}
}
從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
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));消費者最簡單，直接從不帶前綴的list中BLPOP讀取就可以
關于“Redis如何實現延遲隊列”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識，可以關注億速云行業資訊頻道，小編每天都會為大家更新不同的知識點。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
