Redis實(shí)現(xiàn)延遲隊(duì)列的項(xiàng)目示例
最近用到一個延遲消息的功能，第一時間想到使用MQ或者M(jìn)Q的插件，因?yàn)閿?shù)據(jù)量不大，所以嘗試使用Redis來實(shí)現(xiàn)了，畢竟Redis也天生支持類似MQ的隊(duì)列消費(fèi)，所以，在這里總結(jié)了一下Redis實(shí)現(xiàn)延遲消息隊(duì)列的方式。
一、監(jiān)聽key過期時間
處理流程：當(dāng)redis的一個key過期時，redis會生成一個事件，通知訂閱了該事件的客戶端（KeyExpirationEventMessageListener），然后在客戶端的回調(diào)方法中處理邏輯。
1）新建SpringBoot項(xiàng)目，maven依賴及yml如下
maven依賴：
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>
yml文件
server:
port: 8000
spring:
redis:
database: 0
host: xxxx
port: 6379
password: xxxxxx
lettuce:
pool:
#最大連接數(shù)
max-active: 8
#最大阻塞等待時間
max-wait: -1
#最大空閑
max-idle: 8
#最小空閑
min-idle: 0
#連接超時時間
timeout: 5000
2）修改redis.conf文件開啟事件通知配置
默認(rèn)的配置：notify-keyspace-events “”
修改為：notify-keyspace-events Ex，該配置表示監(jiān)聽key的過期事件
3）設(shè)置Redis監(jiān)聽配置，注入Bean RedisMessageListenerContaine
@Configuration
public class RedisTimeoutConfiguration {
@Autowired
private RedisConnectionFactory redisConnectionFactory;
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer() {
RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
return redisMessageListenerContainer;
}
@Bean
public KeyExpiredListener keyExpiredListener() {
return new KeyExpiredListener(this.redisMessageListenerContainer());
}
}
4）創(chuàng)建監(jiān)聽器類，重寫key過期回調(diào)方法onMessage
@Slf4j
public class KeyExpiredListener extends KeyExpirationEventMessageListener {
@Autowired
public RedisTemplate<String, String> redisTemplate;
public KeyExpiredListener(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
@Override
public void onMessage(Message message, byte[] bytes) {
String channel = new String(message.getChannel(), StandardCharsets.UTF_8);
//過期的key
String key = new String(message.getBody(), StandardCharsets.UTF_8);
log.info("redis key 過期：bytes={},channel={},key={}", new String(bytes), channel, key);
}
}
5）編寫測試接口：寫入一個帶過期時間的key
@RestController
@RequestMapping("/demo")
public class BasicController {
@Autowired
public RedisTemplate<String, String> redisTemplate;
@GetMapping(value = "/test")
public void redisTest() {
redisTemplate.opsForValue().set("test", "5s后過期", 5, TimeUnit.SECONDS);
}
}
執(zhí)行后，onMessage監(jiān)聽方法打印結(jié)果:
redis key 過期：bytes=__keyevent@*__:expired,channel=__keyevent@0__:expired,key=test
該方案缺點(diǎn)：可靠性問題，Redis 是一個內(nèi)存數(shù)據(jù)庫，盡管它提供了數(shù)據(jù)持久化選項(xiàng)（如 RDB 和 AOF），但在某些情況下（如意外崩潰或重啟），可能會丟失一些未處理的過期事件。
二、zset + score
基本思路是將消息按需發(fā)送的時間作為分?jǐn)?shù)存儲在有序集合zset中，然后定期檢查并處理到期的消息。代碼例子如下：
1）創(chuàng)建 DelayedMessageService 類
@Slf4j
@Service
public class DelayedMessageService {
private static final String DELAYED_MESSAGES_ZSET = "delayed:messages";
@Autowired
private RedisTemplate<String, String> redisTemplate;
public void addMessage(String message, long delayMillis) {
long score = System.currentTimeMillis() + delayMillis;
redisTemplate.opsForZSet().add(DELAYED_MESSAGES_ZSET, message, score);
}
@Scheduled(fixedRate = 1000)
public void processMessages() {
long now = System.currentTimeMillis();
Set<ZSetOperations.TypedTuple<String>> messages = redisTemplate.opsForZSet().rangeByScoreWithScores(DELAYED_MESSAGES_ZSET, 0, now);
if (messages != null && !messages.isEmpty()) {
for (ZSetOperations.TypedTuple<String> message : messages) {
String msg = message.getValue();
long score = message.getScore().longValue();
if (score <= now) {
// Process the message
System.out.println("Processing message: " + msg);
// Remove the message from the zset
redisTemplate.opsForZSet().remove(DELAYED_MESSAGES_ZSET, msg);
}
}
}else{
log.info("定時任務(wù)執(zhí)行~");
}
}
}
2）編寫Controller接口測試，初始化zset內(nèi)容
@RestController
@RequestMapping("/demo")
public class BasicController {
@Autowired
private DelayedMessageService delayedMessageService;
@GetMapping(value = "/test2")
public void redisZsetTest() {
// Add some messages with delays
delayedMessageService.addMessage("Message 1", 5000); // 5 seconds delay
delayedMessageService.addMessage("Message 2", 10000); // 10 seconds delay
delayedMessageService.addMessage("Message 3", 15000); // 15 seconds delay
}
}
說明：
- redisZsetTest接口通過調(diào)用
DelayedMessageService的addMessage方法，將消息及其到期時間添加到 Redis 的 zset 中 - 開啟一個定時任務(wù)，定期檢查和處理到期的消息。使用
@Scheduled注解定期執(zhí)行，每秒檢查一次，注意這里使用@Scheduled，不要忘了啟動類上添加@EnableScheduling注解，否則定時任務(wù)不會生效。fixedRate屬性表示以固定的頻率（毫秒為單位）執(zhí)行方法。即方法執(zhí)行完成后，會立即等待指定的毫秒數(shù)，然后再次執(zhí)行。 - 通過
redisTemplate.opsForZSet().rangeByScoreWithScores方法按時間范圍獲取到期的消息，消息處理完成后，從zset 中移除處理過的消息
三、Redisson框架
利用 Redisson 提供的數(shù)據(jù)結(jié)構(gòu)RDelayedQueue和RBlockingDeque，可以自動處理過期的任務(wù)并將它們移動到阻塞隊(duì)列中，這樣我們就可以從阻塞隊(duì)列中獲取任務(wù)并進(jìn)行消費(fèi)處理。例子如下：
1）添加依賴
<!-- Redisson 依賴項(xiàng) -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>2.15.1</version>
</dependency>
2）創(chuàng)建DelayedMessageService
@Slf4j
@Service
public class DelayedMessageService {
@Autowired
private RedissonClient redissonClient;
private RBlockingDeque<String> blockingDeque;
private RDelayedQueue<String> delayedQueue;
@PostConstruct
public void init() {
this.blockingDeque = redissonClient.getBlockingDeque("delayedQueue");
this.delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
Executors.newSingleThreadExecutor().submit(this::processMessages);
}
public void addMessage(String message, long delayMillis) {
delayedQueue.offer(message, delayMillis, TimeUnit.MILLISECONDS);
}
public void processMessages() {
try {
while (true) {
String message = blockingDeque.take();
// Process the message
log.info("消息被處理: " + message);
// ..業(yè)務(wù)邏輯處理
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("中斷異常",e);
}
}
}
3）測試接口
@GetMapping(value = "/test3")
public void redisQueueTest() {
// Add some messages with delays
delayedMessageService.addMessage("Message 1", 5000); // 5 seconds delay
delayedMessageService.addMessage("Message 2", 10000); // 10 seconds delay
delayedMessageService.addMessage("Message 3", 15000); // 15 seconds delay
}
說明：
RDelayedQueue 是 Redisson 提供的延遲隊(duì)列，它將消息存儲在指定的隊(duì)列中，直到消息到期才會被轉(zhuǎn)移到該隊(duì)列。它的主要作用包括：
- 延遲消息管理：我們可以使用
RDelayedQueue的offer方法將消息添加到延遲隊(duì)列，并指定延遲時間，消息在延遲時間到期前一直保留在RDelayedQueue中。 - 消息轉(zhuǎn)移：一旦消息到期，
RDelayedQueue會自動將消息轉(zhuǎn)移到指定的RBlockingDeque中。
RBlockingQueue是 Redisson 提供的阻塞隊(duì)列，它支持阻塞操作。主要作用包括：
- 阻塞操作：支持阻塞的
take操作，如果隊(duì)列中沒有元素，會一直阻塞直到有元素可供消費(fèi)。
總結(jié)：
個人推薦使用Redisson 的RDelayedQueue 方式，感覺更加可靠和簡單一些，當(dāng)然zset+score也可以是個不錯選擇，畢竟更加靈活，延遲消息還有其他不同的方案，比如rocketmq、rabbitmq插件等，假如項(xiàng)目中用了redis，又不想引入更多的中間件，可以嘗試使用redis來實(shí)現(xiàn)，為了測試，這里例子都比較簡單，在實(shí)際使用過程中，還要考慮補(bǔ)償機(jī)制、冪等性等問題。
參考：
1.https://blog.csdn.net/qq_34826261/article/details/120598731
到此這篇關(guān)于Redis實(shí)現(xiàn)延遲隊(duì)列的項(xiàng)目示例的文章就介紹到這了,更多相關(guān)Redis 延遲隊(duì)列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Redis的4種緩存模式分享，文章圍繞主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價值，感興趣的小伙伴可以參考一下2022-07-07
這篇文章主要給大家介紹了關(guān)于redis實(shí)現(xiàn)排行榜的簡單方法，文中通過示例代碼介紹的非常詳細(xì)，對大家學(xué)習(xí)或者使用redis具有一定的參考學(xué)習(xí)價值，需要的朋友們下面來一起學(xué)習(xí)學(xué)習(xí)吧2019-08-08
如何通過redis減庫存的秒殺場景實(shí)現(xiàn)
本文通過解決秒殺系統(tǒng)中的一個場景即數(shù)據(jù)預(yù)加載，即把庫存數(shù)據(jù)事先加載到緩存，然后通過緩存來更新庫存，簡單介紹了如何通過redis減庫存的秒殺場景實(shí)現(xiàn)，感興趣的可以了解一下2022-06-06
Redis是一個開源的使用ANSI C語言編寫、支持網(wǎng)絡(luò)、可基于內(nèi)存亦可持久化的日志型、Key-Value數(shù)據(jù)庫，并提供多種語言的API。從2010年3月15日起，Redis的開發(fā)工作由VMware主持。這篇文章主要介紹的是利用yum安裝Redis的方法，有需要的朋友們可以參考借鑒，下面來一起看看吧2016-11-11
Redis恢復(fù)被移除集群的服務(wù)器實(shí)操步驟
這篇文章主要為大家介紹了Redis恢復(fù)被移除集群的服務(wù)器實(shí)操步驟，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-07-07
Redis Sentinel實(shí)現(xiàn)哨兵模式搭建小結(jié)
這篇文章主要介紹了Redis Sentinel實(shí)現(xiàn)哨兵模式搭建小結(jié)，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-12-12
本文詳細(xì)剖析了Redis批量刪除鍵時可能引發(fā)的性能問題及解決方案,通過分析常見的批量刪除方法（如KEYS+DEL、SCAN+DEL、UNLINK）的的優(yōu)缺點(diǎn),提出了改進(jìn)方案,希望幫助大家在實(shí)際業(yè)務(wù)中有效規(guī)避Redis批量刪除鍵的風(fēng)險(xiǎn)2026-05-05
文章介紹Redis中BigKey的概念及其對性能的影響,包括內(nèi)存占用、網(wǎng)絡(luò)阻塞、CPU耗盡和集群不穩(wěn)定等問題,對Redis中BigKey的隱患問題感興趣的朋友一起看看吧2025-07-07
Redis的使用模式之計(jì)數(shù)器模式實(shí)例
這篇文章主要介紹了Redis的使用模式之計(jì)數(shù)器模式實(shí)例,本文講解了匯總計(jì)數(shù)器、按時間匯總的計(jì)數(shù)器、速度控制、使用 Hash 數(shù)據(jù)類型維護(hù)大量計(jì)數(shù)器等內(nèi)容,需要的朋友可以參考下2015-03-03
Redis?哈希Hash底層數(shù)據(jù)結(jié)構(gòu)詳解
這篇文章主要介紹了Redis?哈希Hash底層數(shù)據(jù)結(jié)構(gòu)詳解的相關(guān)資料,需要的朋友可以參考下2022-08-08