SpringBoot集成Redisson实现延迟队列

一、场景

1、下单未支付，超过10分钟取消订单

2、货到后7天未评价，自动好评

二、实现方案

1、使用xxl-job 定时任务按时检测，实时性不高

2、使用RabitMQ的插件rabbitmq_delayed_message_exchange插件

3、 redis的过期检测 redis.conf 中，加入一条配置notify-keyspace-events Ex开启过期监听

等等有很多方法，本文探索SpringBoot+Redisson实现该业务

三、代码

1、pom依赖

<!--        引入 redisson--><dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.15.4</version></dependency>

2、prop配置redis配置

spring:redis:host: 127.0.0.1port: 6379password: redisdatabase: 1timeout: 6000

3、创建RedissionConfig配置

config/RedissonConfig.java

package com.msb.crm.config;import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 创建 RedissonConfig 配置* <p>* Created by fengqx*/
@Configuration
public class RedissonConfig {@Value("${spring.redis.host}")private String host;@Value("${spring.redis.port}")private int port;@Value("${spring.redis.database}")private int database;@Value("${spring.redis.password}")private String password;@Beanpublic RedissonClient redissonClient() {Config config = new Config();config.useSingleServer().setAddress("redis://" + host + ":" + port).setDatabase(database);
//                .setPassword(password);return Redisson.create(config);}
}

4、延迟队列工具类

utils/RedisDelayQueueUtil.java 与

utils/SpringUtil

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.concurrent.TimeUnit;/*** 封装 Redis 延迟队列工具* <p>* Created by fengq*/
@Slf4j
@Component
public class RedisDelayQueueUtil {@Autowiredprivate RedissonClient redissonClient;/*** 添加延迟队列** @param t* @param delay* @param timeUnit* @param queueCode* @param <T>*/public <T> void addDelayQueue(T t, long delay, TimeUnit timeUnit, String queueCode) {try {RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);delayedQueue.offer(t, delay, timeUnit);log.info("添加延时队列成功，队列键：{}，队列值：{}，延迟时间：{}", queueCode, t, timeUnit.toSeconds(delay) + "秒");} catch (Exception e) {log.error("添加延时队列失败：{}", e.getMessage());throw new RuntimeException("添加延时队列失败");}}/*** 获取延迟队列** @param queueCode* @param <T>* @return* @throws InterruptedException*/public <T> T getDelayQueue(String queueCode) throws InterruptedException {RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);T value = (T) blockingDeque.take();return value;}
}

package com.msb.crm.util;import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;/*** SpringUtil 工具类* <p>* Created by fengq*/
@Slf4j
@Component
public class SpringUtil implements ApplicationContextAware {private static ApplicationContext applicationContext;/*** 服务器启动，Spring容器初始化时，当加载了当前类为bean组件后，将会调用下面方法注入ApplicationContext实例** @param applicationContext* @throws BeansException*/@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {log.info("applicationContext 初始化了");SpringUtil.applicationContext = applicationContext;}public static ApplicationContext getApplicationContext() {return applicationContext;}public static <T> T getBean(String beanId) {return (T) applicationContext.getBean(beanId);}public static <T> T getBean(Class<T> clazz) {return (T) applicationContext.getBean(clazz);}
}

5、业务枚举

/entity/RedisDelayQueue.java

/*** 延迟队列业务枚举* <p>* Created by fengq*/
@Getter
@NoArgsConstructor
@AllArgsConstructor
public enum RedisDelayQueue {ORDER_PAYMENT_TIMEOUT("ORDER_PAYMENT_TIMEOUT", "订单支付超时，自动取消订单", "orderPaymentTimeout"),ORDER_TIMEOUT_NOT_EVALUATED("ORDER_TIMEOUT_NOT_EVALUATED", "订单超时未评价，系统默认好评", "orderTimeoutNotEvaluated"),;/*** 延迟队列 Redis Key*/private String code;/*** 中文描述*/private String desc;/*** 延迟队列具体业务实现的 Bean* 可通过 Spring 的上下文获取*/private String beanId;
}

6、延迟对接执行器接口与执行器类

redis/RedisDelayQueueHandle.java 与 redis/RedisDelayQueueRunner.java

package com.msb.crm.redis;/*** 延迟队列执行器* <p>* Created by fenq*/
public interface RedisDelayQueueHandle<T> {void execute(T t);
}

package com.msb.crm.redis;import com.msb.crm.entity.RedisDelayQueue;
import com.msb.crm.util.RedisDelayQueueUtil;
import com.msb.crm.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;/*** 创建延迟队列消费线程* <p>* Created by fenq*/
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {@Autowiredprivate RedisDelayQueueUtil redisDelayQueueUtil;@Autowiredprivate SpringUtil springUtil;/*** 启动延迟队列** @param args*/@Overridepublic void run(String... args) {new Thread(() -> {while (true) {try {RedisDelayQueue[] queues = RedisDelayQueue.values();for (RedisDelayQueue queue : queues) {Object o = redisDelayQueueUtil.getDelayQueue(queue.getCode());if (null != o) {RedisDelayQueueHandle redisDelayQueueHandle = springUtil.getBean(queue.getBeanId());redisDelayQueueHandle.execute(o);}}} catch (InterruptedException e) {log.error("Redis延迟队列异常中断：{}", e.getMessage());}}}).start();log.info("Redis延迟队列启动成功");}
}

6、实现延迟业务-执行方法接口

• OrderPaymentTimeout：订单支付超时延迟队列处理类

package com.msb.crm.redis;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.Map;/*** 订单支付超时处理* <p>* Created by fenq*/
@Slf4j
@Component
public class OrderPaymentTimeout implements RedisDelayQueueHandle<Map<String, Object>> {@Overridepublic void execute(Map<String, Object> map) {// TODO-MICHAEL: 2023-08-05 订单支付超时，自动取消订单处理业务...log.info("收到订单支付超时延迟消息：{}", map);}
}

• OrderTimeoutNotEvaluated：订单超时未评价延迟队列处理类

package com.msb.crm.redis;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.Map;/*** 订单超时未评价处理* <p>* Created by fqngq*/
@Slf4j
@Component
public class OrderTimeoutNotEvaluated implements RedisDelayQueueHandle<Map<String, Object>> {@Overridepublic void execute(Map<String, Object> map) {// TODO-MICHAEL: 2023-08-05 订单超时未评价，系统默认好评处理业务...log.info("收到订单超时未评价延迟消息：{}", map);}
}

7、创建Controller方法

controller/RedisDelayQueueController.java

package com.msb.crm.controller;import com.msb.crm.entity.RedisDelayQueue;
import com.msb.crm.util.RedisDelayQueueUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;/*** 延迟队列测试* <p>* Created by fqngq*/
@RestController
@RequestMapping("/api/redis/delayQueue")
public class RedisDelayQueueController {@Autowiredprivate RedisDelayQueueUtil redisDelayQueueUtil;@GetMapping("/add")public void addQueue() {Map<String, Object> map1 = new HashMap<>();map1.put("orderId", "100");map1.put("remark", "订单支付超时，自动取消订单");Map<String, String> map2 = new HashMap<>();map2.put("orderId", "200");map2.put("remark", "订单超时未评价，系统默认好评");// 添加订单支付超时，自动取消订单延迟队列。为了测试效果，延迟30秒钟redisDelayQueueUtil.addDelayQueue(map1, 30, TimeUnit.SECONDS, RedisDelayQueue.ORDER_PAYMENT_TIMEOUT.getCode());// 订单超时未评价，系统默认好评。为了测试效果，延迟60秒钟redisDelayQueueUtil.addDelayQueue(map2, 60, TimeUnit.SECONDS, RedisDelayQueue.ORDER_TIMEOUT_NOT_EVALUATED.getCode());}
}

四、通过启动该接口，可以复现出延迟队列的执行逻辑

 本人还尝试了，添加队列，然后关闭应用。此时redis数据依旧保留

等一段时间(超过关闭时间)重启项目，此时也不会执行之前队列的数据，需要重新加入数据到队列，在消费新产生队列的后续可以消费到之前的内容(上次项目未执行完毕的)

此时可以打印出上次未执行完毕的数据，因此可以保证数据的最终一致性，可以有效在分布式应用中使用