1、对于一些特殊场景需要延迟消息来改变状态等,比如订单半小时后过期,某个活动在各个时间段用户的动作如果没做就接受消息修改状态,这种方式很实用。采用RDelayedQueue
2、具体原理网上有参数,这是一种拉取redis中zset的数据方式。经过测试很稳定
https://zhuanlan.zhihu.com/p/343811173
3、具体代码
3.1、发送消息
RBlockingDeque<Long> blockingDeque = redissonClient.getBlockingDeque("QUEUE:ORDER");
RDelayedQueue<Long> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
long end = LocalDateTime.parse(endTime, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
long now = LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
if (end > now)
delayedQueue.offer(orderId, end - now, TimeUnit.MILLISECONDS);
3.2、接收消息处理
@Component
@Order(1)
public class StartServerConfig implements ApplicationRunner {
private final static Logger LOGGER = LoggerFactory.getLogger(StartServerConfig.class);
@Resource
private RedissonClient redissonClient;
private final TaskPool executorService = TaskPoolFactory
.getTaskPool(5, 10, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100),
new ThreadPoolExecutor.AbortPolicy());
@Override
public void run(ApplicationArguments args) {
RBlockingDeque<Long> blockingDeque = redissonClient.getBlockingDeque("QUEUE:ORDER");
while (true) {
try {
Long poll = blockingDeque.take();
if (Objects.nonNull(poll))
executorService.submit(new OrderExecutorTask(poll));
} catch (InterruptedException | BusinessException e) {
e.printStackTrace();
//日志平台记录
LOGGER.error(e.getMessage());
}
}
}
}
public class OrderExecutorTask extends ExecutorTask {
private final static Logger LOGGER = LoggerFactory.getLogger(OrderExecutorTask.class);
private Long orderId;
public OrderExecutorTask(Long orderId){
this.orderId = orderId;
}
@Override
public void execute() {
LOGGER.info("=======");
LOGGER.info("订单ID:{}",orderId);
}
}