Redis優(yōu)雅地實現(xiàn)延遲隊列的方法分享
更新時間：2023年02月26日 10:50:08 作者：Hooomeey
Redisson是Redis服務(wù)器上的分布式可伸縮Java數(shù)據(jù)結(jié)構(gòu)，這篇文中主要為大家介紹了Redisson實現(xiàn)的優(yōu)雅的延遲隊列的方法，需要的可以參考一下
前言
工作中常常會遇到這樣的場景，如訂單到期未支付取消，到期自動續(xù)費等，我們發(fā)現(xiàn)延遲隊列非常適合在這樣的場景中使用。常見的延遲隊列的優(yōu)秀實現(xiàn)有rabbitMQ的死信隊列，RocketMQ的延遲隊列等，但是了有時候項目沒有特別的大，沒有引入類似的消息中間件，但是了又遇到了特別適合使用延遲隊列的場景，我們一般會利用已有的redis實現(xiàn)一個簡陋的延遲隊列。常見的實現(xiàn)方式有監(jiān)聽過期key，使用zset利用分值進行一個匹配，但是了這些實現(xiàn)或多或少有些問題，不夠優(yōu)雅。監(jiān)聽過期key是一種危險行為，一是如果過redis中key數(shù)量較大監(jiān)聽過期key可能導(dǎo)致服務(wù)負載異常，二是redis中key過期后key是惰性刪除的，因此監(jiān)聽機制需要主動觸發(fā)。利用zset分值實現(xiàn)呢，需要自己開發(fā)代碼處理定時輪訓(xùn)以及key刪除的邏輯，具有一定的工作量和復(fù)雜度。哪有沒有一種優(yōu)雅的redis延遲隊列的實現(xiàn)呢？
Redisson是Redis服務(wù)器上的分布式可伸縮Java數(shù)據(jù)結(jié)構(gòu)----駐內(nèi)存數(shù)據(jù)網(wǎng)格(In-Memory Data Grid，IMDG)。底層使用netty框架，并提供了與java對象相對應(yīng)的分布式對象、分布式集合、分布式鎖和同步器、分布式服務(wù)等一系列的Redisson的分布式對象。為我們提供了許多開箱即用的功能。今天介紹Redisson實現(xiàn)的優(yōu)雅的延遲隊列。
使用
依賴配置
<?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>配置文件
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
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);
}
}
}執(zhí)行效果
原理分析
從RedissonDelayedQueue實現(xiàn)中我們看到有四個角色
- redisson_delay_queue_timeout:xxx，sorted set數(shù)據(jù)類型，存放所有延遲任務(wù)，按照延遲任務(wù)的到期時間戳（提交任務(wù)時的時間戳 + 延遲時間）來排序的，所以列表的最前面的第一個元素就是整個延遲隊列中最早要被執(zhí)行的任務(wù)，這個概念很重要
- redisson_delay_queue:xxx，list數(shù)據(jù)類型，暫時沒發(fā)現(xiàn)什么用，只是在提交任務(wù)時會寫入這里面，隊列轉(zhuǎn)移時又會刪除里面的元素
- xxx：list數(shù)據(jù)類型，被稱為目標隊列，這個里面存放的任務(wù)都是已經(jīng)到了延遲時間的，可以被消費者獲取的任務(wù)，所以上面demo中的RBlockingQueue的take方法是從這個目標隊列中獲取到任務(wù)的
- redisson_delay_queue_channel:xxx，是一個channel，用來通知客戶端開啟一個延遲任務(wù)
隊列創(chuàng)建
RedissonDelayedQueue延遲隊列創(chuàng)建時，指定了隊列轉(zhuǎn)移服務(wù)，以及實現(xiàn)延遲隊列的四個重要校色的key。核心代碼是指定隊列轉(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]); "http://拿到zset中過期的值列表
+ "if #expiredValues > 0 then " //如果有
+ "for i, v in ipairs(expiredValues) do "
+ "local randomId, value = struct.unpack('dLc0', v);"http://解構(gòu)消息，在提交任務(wù)時打包的消息
+ "redis.call('rpush', KEYS[1], value);" //放入無前綴的list 隊頭
+ "redis.call('lrem', KEYS[3], 1, v);"http://移除帶前綴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'); "http://取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);
}
};生產(chǎn)者
核心代碼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);"http://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); "http://取出zset中第一個元素
+ "if v[1] == value then " //如果最快過期的元素就是這次發(fā)送的消息
+ "redis.call('publish', KEYS[4], ARGV[1]); " //channel中發(fā)布一下超時時間
+ "end;",
Arrays.asList(getRawName(), timeoutSetName, queueName, channelName),
timeout, randomId, encode(e));消費者
消費者最簡單，直接從不帶前綴的list中BLPOP讀取就可以
整個流程
總結(jié)思考
Lua是redis的好朋友，我們可以看到Redisson實現(xiàn)延遲隊列時，大量使用到lua腳本，因Redis會將整個腳本作為一個整體執(zhí)行，中間不會被其他請求插入。因此在腳本運行過程中無需擔(dān)心會出現(xiàn)競態(tài)條件，無需使用事務(wù)。我們在平時開發(fā)時有多個redis命令操作的有簡單的業(yè)務(wù)邏輯，不妨嘗試一下lua腳本的方式，可以避免使用分布式鎖來保障一致性。
Redisson的源碼值得一讀，有很多新東西值得學(xué)習(xí)，如果其用到的netty基于時間輪算法的定時任務(wù)調(diào)度，可以讓我們基于此實現(xiàn)自己的任務(wù)調(diào)度框架，也讓我有了去探究這種實現(xiàn)方式和基于ScheduledThreadPoolExecutor的定時調(diào)度的差異及各自優(yōu)劣的欲望。
以上就是Redis優(yōu)雅地實現(xiàn)延遲隊列的方法分享的詳細內(nèi)容，更多關(guān)于Redis延遲隊列的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
雖說Redis是內(nèi)存數(shù)據(jù)庫，但是它為數(shù)據(jù)的持久化提供了兩個技術(shù)，分別是AOF日志和RDB快照。這兩種技術(shù)都會用各用一個日志文件來記錄信息，但是記錄的內(nèi)容是不同的。AOF 文件的內(nèi)容是操作命令； RDB 文件的內(nèi)容是二進制數(shù)據(jù)。本文將討論RDB快照的原理和使用2021-06-06
Redis管道操作通過將多個命令一次性發(fā)送到服務(wù)器,減少了網(wǎng)絡(luò)往返次數(shù),本文就來介紹一下Redis的管道操作,具有一定的參考價值,感興趣的可以了解一下2025-02-02
這篇文章主要介紹了Redis Key過期策略的用法,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-06-06
在很多場景下,我們需要對大量的數(shù)據(jù)進行分頁展示,本文主要介紹了使用redis實現(xiàn)高效分頁的項目實踐,具有一定的參考價值,感興趣的可以了解一下2024-02-02
這篇文章主要介紹了詳解redis中的鎖以及使用場景，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-12-12
Redis 的 HyperLogLog 是一種概率性數(shù)據(jù)結(jié)構(gòu),用于統(tǒng)計唯一元素的數(shù)量（基數(shù)）,本文主要介紹了Redis中HyperLogLog的使用小結(jié),感興趣的可以了解一下2025-05-05
Redis 操作多個數(shù)據(jù)庫的配置的方法實現(xiàn)
本文主要介紹了Redis 操作多個數(shù)據(jù)庫的配置的方法實現(xiàn)，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-03-03
Redis的分布式鎖是通過利用Redis的原子操作和特性來實現(xiàn)的,為了保證數(shù)據(jù)的一致性和避免沖突,可以使用分布式鎖來進行同步控制,本文給大家介紹了如何解決Redis分布式鎖的誤刪問題和原子性問題,需要的朋友可以參考下2024-02-02
這篇文章主要介紹了使用Grafana監(jiān)控Redis,號稱下一代可視化監(jiān)控系統(tǒng)，結(jié)合SpringBoot使用，本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-04-04
本文主要介紹了Redis分布式鎖的幾種實現(xiàn)方法,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2025-04-04