赞
踩
对于延时比较长的场景、实时性不高的场景,我们可以采用任务调度的方式定时轮询处理。如:xxl-job
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.15.4</version>
</dependency>
spring:
redis:
host: 127.0.0.1
port: 6379
password: 123456
database: 1
timeout: 6000
/** * 创建 RedissonConfig 配置 * <p> * Created by michael on 2021-04-25. */ @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); } }
/** * 封装 Redis 延迟队列工具 * <p> * Created by michael on 2021-04-25. */ @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; } }
/** * SpringUtil 工具类 * <p> * Created by michael on 2021-04-25. */ @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); } }
/** * 延迟队列业务枚举 * <p> * Created by michael on 2021-04-25. */ @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; }
/**
* 延迟队列执行器
* <p>
* Created by michael on 2021-04-25.
*/
public interface RedisDelayQueueHandle<T> {
void execute(T t);
}
/**
* 订单支付超时处理
* <p>
* Created by michael on 2021-04-25.
*/
@Slf4j
@Component
public class OrderPaymentTimeout implements RedisDelayQueueHandle<Map<String, Object>> {
@Override
public void execute(Map<String, Object> map) {
// TODO-MICHAEL: 2021-04-25 订单支付超时,自动取消订单处理业务...
log.info("收到订单支付超时延迟消息:{}", map);
}
}
/**
* 订单超时未评价处理
* <p>
* Created by michael on 2021-04-25.
*/
@Slf4j
@Component
public class OrderTimeoutNotEvaluated implements RedisDelayQueueHandle<Map<String, Object>> {
@Override
public void execute(Map<String, Object> map) {
// TODO-MICHAEL: 2021-04-25 订单超时未评价,系统默认好评处理业务...
log.info("收到订单超时未评价延迟消息:{}", map);
}
}
/** * 创建延迟队列消费线程 * <p> * Created by michael on 2021-04-25. */ @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延迟队列启动成功"); } }
以上步骤,Redis 延迟队列核心代码已经完成,下面写一个测试接口,用 PostMan 模拟测试一下
/** * 延迟队列测试 * <p> * Created by michael on 2021-04-25. */ @RestController @RequestMapping("/api/redis/delayQueue") public class RedisDelayQueueController { @Autowired private RedisDelayQueueUtil redisDelayQueueUtil; @PostMapping("/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", "订单超时未评价,系统默认好评"); // 添加订单支付超时,自动取消订单延迟队列。为了测试效果,延迟10秒钟 redisDelayQueueUtil.addDelayQueue(map1, 10, TimeUnit.SECONDS, RedisDelayQueue.ORDER_PAYMENT_TIMEOUT.getCode()); // 订单超时未评价,系统默认好评。为了测试效果,延迟20秒钟 redisDelayQueueUtil.addDelayQueue(map2, 20, TimeUnit.SECONDS, RedisDelayQueue.ORDER_TIMEOUT_NOT_EVALUATED.getCode()); } }
通过 Redis 客户端可看到两个延迟队列已添加成功
查看 IDEA 控制台日志可看到延迟队列已消费成功
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。