从0~1实现单体或微服务下实现订单未支付超时取消功能方案（2）-rocketmq 延迟队列方案完整设计和源码

news2025/2/22 13:00:48

从0~1实现单体或微服务下订单未支付超时取消功能方案（1）-java delayquene + 注册中心(zookeeper/nacos)高可用方案
从0~1实现单体或微服务下订单未支付超时取消功能方案（2）-rocketmq 延迟队列方案

场景说明

我们日常接触的电商或者票类平台等都有一些共通的功能，比如下单了，超过某时间未支付，则会自动取消该订单开发的角度来说就是订单从创建成功状态到取消状态，这个并不能靠前端去维护，一个是安全角度不够，另一个就是性能不足和维护复杂度太高
后台去处理就涉及了同步取消和异步取消两大种类方式，具体如下
说明：本文基于minicloud框架下，各位可以用各自得微服务项目即可，本文主要是提供其中一种思路

常用的实现方案以及优缺点

同步方式-查询时再去同时根绝订单创建时间更取消订单

优点：开发简单
缺点：无法实时，而且比较耗时耗性能，并发性低

异步方式-定时任务轮训取消订单

优点：开发简单，可做高可用
缺点：实时性差，瞬间高并发（比如秒杀）下定时的时间很难做到适中

异步方式-redis key过期通知，需要提前防止一个订单id 对应key以及对应过期时间，过期会触发通知

优点：可以做到实时，复杂度低
缺点：redis 订阅通知不保证一定达到消费端，即使消费端都宕机了也不会重复发送

异步方式-java delayquene

优点： java原生的阻塞队列 delayquene，开发便捷
缺点：内存队列，宕机队列既消失，无法高可用（需配合一些第三方进行高可用，本文是配合nacos注册中心）

异步方式-rocketmq 延时队列

优点：高可用
缺点：但是需要引入mq组件复，杂度增加，开源版仅支持部分固定时间，专业版支持自定义时间
本篇基于rocketmq 延迟队列实现

rocketmq 延迟队列方案介绍

rocketmq 本身支持直接发送延迟消息，发送到broker 后，内部计算时间达到指定时间时才会被消费者可见开源版只支持固定的时间，发送时用level 一对一等价，规则如下：
1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
对应1~18 这18个level,也就是说1对应1s,2对应5s,3对应10s

rocketmq 延迟队列订单取消流程图

在这里插入图片描述

流程说明

如上图所示，流程很简单
1.请求订单服务下单，下单成功后，发送orderId等信息到延迟队列
2.延迟时间到，消费端接到消息
3.调用订单服务更新订单状态，如果订单为未支付，则变更为取消状态

完整源码

共同部分源码

共通源码参考《单体或微服务下实现订单未支付超时取消功能方案（1）-java delayquene + 注册中心(zookeeper/nacos)高可用方案》源码部分

rocketmq 相关发送以及消费端部分源码

发送端代码

在这里插入图片描述

 /**
     * 创建状态为未支付的订单，并发送延迟队列消息
     * */
    @Override
    public void createRandomTimeOrderWithoutPay() {

        //创建订单并插入到数据库
        long orderId = GuidUtil.longGuid();
        OrderEntity testOrder = new OrderEntity();
        testOrder.setOrderId(orderId);
        testOrder.setStatus(1);
        testOrder.setCreateTime(LocalDateTime.now());
        testOrder.setDetail(JSONUtil.toJsonPrettyStr(testOrder));
        orderMapper.insert(testOrder);

        //发送延迟队列
        //clusterServiceInvoker.orderInDelayQuene("core-processer-biz","default",orderId,1);

        //使用发送延迟消息接口发送
        orderWithoutPayQuene.push(orderId,1,TimeUnit.MINUTES);

    }

/**
 * @Author alan.wang
 */
@AllArgsConstructor
@Service("mqOrderWithoutPayQuene")
public class MQOrderWithoutPayQuene implements OrderWithoutPayQuene {

    private final MqMessageSender mqMessageSender;

    @Override
    public void push(long orderId, long expireTime, TimeUnit timeUnit) {

        DelayQueneOrderDTO delayQueneOrderDTO = new DelayQueneOrderDTO();
        delayQueneOrderDTO.setExpireTime(expireTime);
        delayQueneOrderDTO.setOrderId(orderId);
        mqMessageSender.sendOrderWithoutPayInDelayQueneMsg(delayQueneOrderDTO);

    }

    @Override
    public void take() {
       //nothing need to do
    }
}

消费端代码

新建OrderWithoutPayCannelConsumer 消费者，专门处理订单超时未支付延迟队列topic

/**
 * @Author alan.wang
 * 处理订单延迟队列类，将超时未支付的订单状态变为取消
 */
@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = "topic-order-without-pay", consumerGroup = "test_order_without_pay_group")
public class OrderWithoutPayCannelConsumer implements RocketMQListener<String> {


    @Override
    public void onMessage(String delayQueneOrderJson) {
        
        //获取RemoteSimulateOrderService feign 接口
        RemoteSimulateOrderService remoteSimulateOrderService = SpringContextUtils.getBean(RemoteSimulateOrderService.class);
        
        //转换delayQueneOrderJson
        DelayQueneOrderDTO delayQueneOrderDTO = JSONUtil.toBean(delayQueneOrderJson, DelayQueneOrderDTO.class);
        
        //执行变更订单状态为取消的操作
        remoteSimulateOrderService.updateStatusToCancel(delayQueneOrderDTO.getOrderId());
        System.out.println(delayQueneOrderJson+":"+ LocalDateTime.now());
    }


}

MqMessageSender

 /**
    * 发送订单号等延迟队列消息,10s 后触发消费端可见
    */
    public void sendOrderWithoutPayInDelayQueneMsg(DelayQueneOrderDTO delayQueneOrderDTO) {

        Message<String> msg = MessageBuilder.withPayload(JSONUtil.toJsonStr(delayQueneOrderDTO)).build();

        //30000 代表超时时间，3为level 既：10s
        rocketMQTemplate.syncSend(TOPIC_ORDER_WITHOUT_PAY_DESTINATION,msg,30000,3);
    }