您好,登錄后才能下訂單哦!
這篇文章主要介紹“Redis如何實(shí)現(xiàn)延遲隊(duì)列”的相關(guān)知識(shí),小編通過實(shí)際案例向大家展示操作過程,操作方法簡單快捷,實(shí)用性強(qiáng),希望這篇“Redis如何實(shí)現(xiàn)延遲隊(duì)列”文章能幫助大家解決問題。
<?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的自動(dòng)配置[最簡單]
第二種:使用單獨(dú)的redisson配置文件
第三種:使用spring.redis.redisson這個(gè)配置key下進(jìn)行配置
詳細(xì)的整合查看 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; /** * 明天的你會(huì)因今天到的努力而幸運(yùn) * * @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
實(shí)現(xiàn)中我們看到有四個(gè)角色
redisson_delay_queue_timeout:xxx,sorted set數(shù)據(jù)類型,存放所有延遲任務(wù),按照延遲任務(wù)的到期時(shí)間戳(提交任務(wù)時(shí)的時(shí)間戳 + 延遲時(shí)間)來排序的,所以列表的最前面的第一個(gè)元素就是整個(gè)延遲隊(duì)列中最早要被執(zhí)行的任務(wù),這個(gè)概念很重要
redisson_delay_queue:xxx,list數(shù)據(jù)類型,暫時(shí)沒發(fā)現(xiàn)什么用,只是在提交任務(wù)時(shí)會(huì)寫入這里面,隊(duì)列轉(zhuǎn)移時(shí)又會(huì)刪除里面的元素
xxx:list數(shù)據(jù)類型,被稱為目標(biāo)隊(duì)列,這個(gè)里面存放的任務(wù)都是已經(jīng)到了延遲時(shí)間的,可以被消費(fèi)者獲取的任務(wù),所以上面demo中的RBlockingQueue的take方法是從這個(gè)目標(biāo)隊(duì)列中獲取到任務(wù)的
redisson_delay_queue_channel:xxx,是一個(gè)channel,用來通知客戶端開啟一個(gè)延遲任務(wù)
RedissonDelayedQueue
延遲隊(duì)列創(chuàng)建時(shí),指定了隊(duì)列轉(zhuǎn)移服務(wù),以及實(shí)現(xiàn)延遲隊(duì)列的四個(gè)重要校色的key。核心代碼是指定隊(duì)列轉(zhuǎn)移任務(wù)
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);"//解構(gòu)消息,在提交任務(wù)時(shí)打包的消息 + "redis.call('rpush', KEYS[1], value);" //放入無前綴的list 隊(duì)頭 + "redis.call('lrem', KEYS[3], 1, v);"//移除帶前綴list 隊(duì)尾元素 + "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]; " //返回分值,即過期時(shí)間 + "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中加入消息及其超時(shí)分值 + "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中第一個(gè)元素 + "if v[1] == value then " //如果最快過期的元素就是這次發(fā)送的消息 + "redis.call('publish', KEYS[4], ARGV[1]); " //channel中發(fā)布一下超時(shí)時(shí)間 + "end;", Arrays.asList(getRawName(), timeoutSetName, queueName, channelName), timeout, randomId, encode(e));
消費(fèi)者最簡單,直接從不帶前綴的list中BLPOP讀取就可以
關(guān)于“Redis如何實(shí)現(xiàn)延遲隊(duì)列”的內(nèi)容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí),可以關(guān)注億速云行業(yè)資訊頻道,小編每天都會(huì)為大家更新不同的知識(shí)點(diǎn)。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。