Spring Boot集成Redisson实现延迟队列

项目场景：

   在电商、支付等领域，往往会有这样的场景，用户下单后放弃支付了，那这笔订单会在指定的时间段后进行关闭操作，细心的你一定发现了像某宝、某东都有这样的逻辑，而且时间很准确，误差在1s内；那他们是怎么实现的呢？

   一般实现的方法有几种：使用 redisson、rocketmq、rabbitmq等消息队列的延时投递功能。

解决方案：

   一般项目集成redis的比较多，所以我这篇文章就说下redisson延迟队列，如果使用rocketmq或rabbitmq需要额外集成中间件，比较麻烦一点。

1.集成redisson

maven依赖

	org.redisson
	redisson-spring-boot-starter
	3.21.1

yml配置，单节点配置可以兼容redis的配置方式

# redis配置
spring:
  redis:
    database: 0
    host: 127.0.0.1
    password: redis@pass
    port: 6001

 更详细的配置参考：Spring Boot整合Redisson的两种方式-CSDN博客

2.配置多线程

因为延迟队列可能会多个任务同时执行，所以需要多线程处理。

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
import java.util.concurrent.ThreadPoolExecutor;
 
@Configuration
@EnableAsync
public class ExecutorConfig {
    /**
     * 异步任务自定义线程池
     */
    @Bean(name = "taskExecutor")
    public ThreadPoolTaskExecutor asyncServiceExecutor() {
    	ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(50);
        //配置最大线程数
        executor.setMaxPoolSize(500);
        //配置队列大小
        executor.setQueueCapacity(300);
        //允许线程空闲时间
        executor.setKeepAliveSeconds(60);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix("taskExecutor-");
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //调用shutdown()方法时等待所有的任务完成后再关闭
        executor.setWaitForTasksToCompleteOnShutdown(true);
        //等待所有任务完成后的最大等待时间
		executor.setAwaitTerminationSeconds(60);
        return executor;
    }
}

3.具体业务

比如消息通知、关闭订单等 ，这里加上了@Async注解，可以异步执行

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
 
import java.text.SimpleDateFormat;
import java.util.Date;
 
@Service
public class AsyncService {
 
	@Async
	public void executeQueue(Object value) {
		System.out.println();
		System.out.println("当前线程："+Thread.currentThread().getName());
		System.out.println("执行任务："+value);
 
		//打印时间方便查看
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		System.out.println("执行任务的时间："+sdf.format(new Date()));
		//自己的业务逻辑，可以根据id发送通知消息等
		//......
	}
}

4.延迟队列(关键代码)

这里包括添加延迟队列，和消费延迟队列，@PostConstruct注解的意思是服务启动加载一次，参考

Spring Boot项目启动时执行指定的方法-CSDN博客Spring Boot中多个PostConstruct注解执行顺序控制_多个postconstruct执行顺序-CSDN博客

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
 
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;
 
@Service
public class TestService {
 
	@Resource
	private AsyncService asyncService;
	@Resource
	private ThreadPoolTaskExecutor executor;
	@Autowired
	private RedissonClient redissonClient;
 
	/**
	 * 添加延迟任务
	 */
	public void addQueue() {
		//获取延迟队列
		RBlockingQueue