一、场景
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.1
port: 6379
password: redis
database: 1
timeout: 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;
@Bean
public 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 {
@Autowired
private 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 */ @Override public 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 {
@Autowired
private RedisDelayQueueUtil redisDelayQueueUtil;
@Autowired
private SpringUtil springUtil;
/**
* 启动延迟队列
*
* @param args
*/
@Override
public 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>> {
@Override
public 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>> {
@Override
public 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 {
@Autowired
private 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数据依旧保留
等一段时间(超过关闭时间)重启项目，此时也不会执行之前队列的数据，需要重新加入数据到队列，在消费新产生队列的后续可以消费到之前的内容(上次项目未执行完毕的)
此时可以打印出上次未执行完毕的数据，因此可以保证数据的最终一致性，可以有效在分布式应用中使用
