在springboot中利用Redis实现延迟队列

---

前言

在开发过程中，有很多场景都需要用到延迟队列来解决。目前支持延迟队列的中间件也不少，特别是基于JMS模式下的消息中间件基本上都支持延迟队列。但是有时我们项目规模可能比较小，用不上JMS这些中间件。那么利用Redis也可以实现延迟队列的功能。

---

一、基本思路

利用Redis来实现延迟队列的主要思路是借助Redis的Sorted Set数据类型来实现。

具体做法是将任务的执行时间作为分数（score），任务的内容作为值（value），将任务按照执行时间排序存储在有序集合中。然后周期性地检查有序集合中的任务，根据当前时间和任务的执行时间来决定是否执行任务。

当需要添加新的延迟任务时，只需将任务的执行时间和内容添加到有序集合中即可。当然，你可能需要一个后台进程或定时任务来不断地检查有序集合，以执行到期的任务。

二、springboot实现案例

根据上面的思路，我们可以直接来写代码，本案例的完整代码点击下载。

首先，确保你的Spring Boot项目中已经配置好了Redis依赖。你可以在pom.xml文件中添加如下依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

然后，创建一个延迟队列管理器（DelayQueueManager）类，用于添加任务到延迟队列和处理到期任务：

package com.test.spring.redisdelayqueue;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import java.util.Set;
/*** @author code-long* @version 1.0.0* @ClassName DelayQueueManager.java* @description*/
@Component
public class DelayQueueManager {private static final String key = "delayQueue";@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void addToDelayQueue(String value, long delayMillis) {redisTemplate.opsForZSet().add(key, value, System.currentTimeMillis() + delayMillis);}public Set<String> getExpiredItems(long currentTime) {return redisTemplate.opsForZSet().rangeByScore(key, 0, currentTime);}public void removeItems(Set<String> items) {redisTemplate.opsForZSet().remove(key, items.toArray());}
}

接下来，我们利用spring的定时任务创建一个定时任务，用于定期检查延迟队列中的到期任务并执行。
同时我们还需要单独创建一个线程来专门处理逻辑，因为如果在定时任务直接处理逻辑可能会导致定时任务阻塞的现象，在这个线程中我们为了保证队列的顺序性，在使用BlockingDeque来模拟一个队列。当然如果你的队列逻辑处理不需要保持顺序性，完全可以使用多线程来处理任务。
具体实现代码：

package com.test.spring.redisdelayqueue;import lombok.extern.slf4j.Slf4j;
import org.h2.util.DateTimeUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.text.DateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;/*** @author code-long* @version 1.0.0* @ClassName RedisDelayQueueApplication.java* @description*/
@SpringBootApplication(scanBasePackages = "com.test.spring.redisdelayqueue")
@EnableScheduling
@Slf4j
@RestController
public class RedisDelayQueueApplication {@Autowiredprivate DelayQueueManager delayQueueManager;@Scheduled(fixedRate = 1000) // 每秒执行一次public void processDelayQueue() {long currentTime = System.currentTimeMillis();Set<String> expiredItems = delayQueueManager.getExpiredItems(currentTime);// 处理到期任务,这里就可以达到延迟队列的模型//如果在这里直接处理逻辑，会影响到定时任务执行不完全的现象，比如一个任务执行需要2秒，那么就会阻塞JOB的执行，所以我们要另外启动一个线程来专门处理逻辑for (String item : expiredItems) {//将过期数据加入到执行队列DelayQueueInstance.getInstance().receive(item);}// 从延迟队列中移除已处理的任务：这里的删除可以放到线程中逻辑执行完成再删除if(!expiredItems.isEmpty()){delayQueueManager.removeItems(expiredItems);}}//应用启动成功后，就启动线程@EventListenervoid listener(ApplicationReadyEvent event) {DelayQueueInstance.getInstance().start();}//模拟入队操作@RequestMapping("/push")@Transactionalpublic String test1(){//模拟一个30秒的延迟队列delayQueueManager.addToDelayQueue("{这里可以使json数据:30}",30000);delayQueueManager.addToDelayQueue("{这里可以使json数据:10}",10000);System.out.println("["+ LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) +"]添加数据到队列");return "success";}public static void main(String[] args) {SpringApplication.run(RedisDelayQueueApplication.class, args);}
}

三、测试

我们在浏览器中访问或者使用curl调用接口：

curl http://localhost:8881/push

后台打印结果为：

[2024-03-14 22:28:54]添加数据到队列
[2024-03-14 22:29:05]收到数据---{这里可以使json数据:10}
[2024-03-14 22:29:25]收到数据---{这里可以使json数据:30}

我们可以看到，基本上能实现延迟队列的功能，只是这里有一点小小的瑕疵，任务可能会存在1秒的误差，但是这依赖于我们定时任务的循环时间，如果时间越短，误差的时间也就越短，定时任务间隔时间越长，误差也就越大。但1秒误差在实际的业务过程中已经是可以接受的了，对服务器来说性能也可以接受。

---

总结

使用Redis实现延迟队列的好处包括简单、高效，并且Redis本身就具有持久化和高可用性的特性，使得延迟队列的实现更加可靠。如果项目没有必要上JMS中间件，那么使用Redis是一个不错的方案。