电商系统如何实现订单超时自动取消？

朴灿烈づ我的快乐病毒、 2022-10-21 11:52 356阅读 0赞

### 一、背景 ###

系统中用户下单，对于系统下单一般是分布式事务的操作，想要实现订单超时自动取消，我们可以基于MQ的延迟队列和死信队列实现。整体的实现思路分三种情况要考虑，第一种是订单的创建和投递到MQ，第二种是正常订单消息的消费，另外则是超时后消息的消费。

### 二、实现思路 ###

对于订单的创建，只要生产者将消息成功投递到MQ，则认为订单创建成功。MQ返回ack表明消息投递成功，此时向延迟队列发送一条消息，而延迟队列挂载死信队列。这样做目的是：如果延迟队列中的消息达到阈值还没消费，则会进入死信队列，此时死信队列的监听器则会获取到过期的订单信息，可以做取消操作，反之，则走正常订单消费的流程。

整体实现思路大体如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hzZjE1NzY4NjE1Mjg0_size_16_color_FFFFFF_t_70_pic_center]

### 三、具体代码 ###

本文基于RabbitMQ实现，借助于RabbitMQ的延迟队列TTL和死信队列。  
配置文件：

server.port=8080
    
    # 设计rabbitmq连接
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=root
    spring.rabbitmq.password=123456
    # 设置虚拟主机
    spring.rabbitmq.virtual-host=keduw-order
    
    # 设置发布者确认机制
    # correlated发布消息成功到交换器后会触发回调方法，默认是none
    spring.rabbitmq.publisher-confirm-type=correlated
    spring.rabbitmq.publisher-returns=true
    
    # 消息为手动确认
    spring.rabbitmq.listener.direct.acknowledge-mode=manual

增加RabbitMQ的配置类，创建对应的队列、转换器、监听器以及队列信息绑定，备注很详细，这里就不太赘述。

**
     * RabbitMQ配置类
     */
    @Configuration
    public class RabbitMqConfig {
    
        /**
         * 使用DirectMessageListenerContainer，您需要确保ConnectionFactory配置了一个任务执行器，
         * 该执行器在使用该ConnectionFactory的所有侦听器容器中具有足够的线程来支持所需的并发性。
         * 默认连接池大小仅为5。
         *
         * 并发性基于配置的队列和consumersPerQueue。每个队列的每个使用者使用一个单独的通道，
         * 并发性由rabbit客户端库控制;默认情况下，它使用5个线程池;
         * 可以配置taskExecutor来提供所需的最大并发性。
         *
         * @param connectionFactory
         * @return
         */
        @Bean(name = "rabbitMessageListenerContainer")
        public DirectMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory){
            // 写的时候，默认使用DEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2个线程
            DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
            // 设置确认消息的模式
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            container.setPrefetchCount(5);
            container.setConsumersPerQueue(5);
            container.setMessagesPerAck(1);
    
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            taskExecutor.setCorePoolSize(10);
            taskExecutor.setMaxPoolSize(20);
            //设置该属性，灵活设置并发 ,多线程运行。
            container.setTaskExecutor(taskExecutor);
    
            return container;
        }
    
        /**
         * 设置消息转换器，用于将对象转换成JSON数据
         * 可以通过converterAndSend将对象发送消息队列
         * 监听器也可以通过该工具将接受对象反序列化成java对象
         *
         * @return Jackson转换器
         */
        @Bean
        public MessageConverter messageConverter(){
            return new Jackson2JsonMessageConverter();
        }
    
        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
            return new RabbitAdmin(connectionFactory);
        }
    
        /**
         * 订单消息队列
         * @return
         */
        @Bean
        public Queue orderQueue(){
            return QueueBuilder.durable("q.order").build();
        }
    
        /**
         * 延迟消息队列
         * @return
         */
        @Bean
        public Queue ttlQueue(){
            Map<String,Object> args = new HashMap<>();
            // 该队列的消息10s到期
            args.put("x-message-ttl", 10000);
            // 设置死信队列交换器,（当队列消息TTL到期后依然没有消费，则加入死信队列）
            args.put("x-dead-letter-exchange","x.dlx");
            // 设置私信队列路由键,设置该队列所关联的死信交换器的routingKey，如果没有特殊指定，使用原队列的routingKey
            args.put("x-dead-letter-routing-key","k.dlx");
            Queue queue = new Queue("q.ttl",true,false,false, args);
            return queue;
        }
    
        /**
         * 死信队列，用于取消用户订单
         * 当10s还没有付款的订单则进入死信队列，消费死信队列，取消用户订单
         *
         * @return
         */
        @Bean
        public Queue dlxQueue(){
            Map<String,Object> args = new HashMap<>();
            Queue dlq = new Queue("q.dlx",true,false,false, args);
    
            return dlq;
        }
    
        /**
         * 订单交换器
         * @return
         */
        @Bean
        public Exchange orderExchange(){
            Map<String, Object> args = new HashMap<>();
            DirectExchange exchange = new DirectExchange("x.order", true, false, args);
    
            return exchange;
        }
    
        /**
         * 延迟队列交换器
         * @return
         */
        @Bean
        public Exchange ttlExchange(){
            Map<String, Object> args = new HashMap<>();
            return new DirectExchange("x.ttl", true, false, args);
        }
    
        /**
         * 死信队列交换器
         * @return
         */
        @Bean
        public Exchange dlxExchange(){
            Map<String, Object> args = new HashMap<>();
            DirectExchange exchange = new DirectExchange("x.dlx", true, false, args);
            return exchange;
        }
    
        /**
         * 用于发送下单，做分布式事务的MQ
         * @return
         */
        @Bean
        public Binding orderBinding(){
            return BindingBuilder.bind(orderQueue())
                    .to(orderExchange())
                    .with("k.order")
                    .noargs();
        }
    
        /**
         * 用于等待用户支付的延迟队列绑定
         * @return
         */
        @Bean
        public Binding ttlBinding(){
            return BindingBuilder.bind(ttlQueue())
                    .to(ttlExchange())
                    .with("k.ttl")
                    .noargs();
        }
    
        /**
         * 用于支付超时取消用户订单的死信队列绑定
         * @return
         */
        @Bean
        public Binding dlxBinding(){
            return BindingBuilder.bind(dlxQueue())
                    .to(dlxExchange())
                    .with("k.dlx")
                    .noargs();
        }
    
    }

创建订单监听器，用于监听订单正常的支付提交和超时取消。

/**
     * 订单正常支付流程监听
     */
    @Component
    public class OrderNormalListener {
    
        @RabbitListener(queues = "q.order",ackMode = "MANUAL")
        public void onMessage(Order order , Channel channel , Message message) throws IOException {
            System.out.println("写入数据库");
            System.out.println(order);
    
            for (OrderDetail detail : order.getDetails()){
                System.out.println(detail);
            }
    
            channel.basicAck(message.getMessageProperties().getDeliveryTag() , false);
        }
    
    }

创建订单超时自动取消监听器，监听的是死信队列。

/**
     * 订单超时自动取消监听
     */
    @Component
    public class OrderCancelListener implements ChannelAwareMessageListener {
    
        @Override
        @RabbitListener(queues = "q.dlx" , ackMode = "MANUAL")
        public void onMessage(Message message, Channel channel) throws Exception {
            String orderId = new String(message.getBody());
            System.out.println("取消订单：" + orderId);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

对于订单提交，正常提交后同时投递多一份到延迟队列里面去，用作延迟取消。

// 构建订单信息
    Order order = new Order();
    order.setUserId(IdUtils.generateUserId());
    order.setOrderId(IdUtils.generateOrderId());
    // 设置状态为待支付
    order.setStatus(OrderStatus.TO_BE_PAYED.toString());
    order.setDetails(details);
    
    // 投递消息
    CorrelationData correlationData = new CorrelationData();
    rabbitTemplate.convertAndSend("x.order","k.order", order, correlationData);
    // 同步等待，可以设置为异步回调
    CorrelationData.Confirm confirm = correlationData.getFuture().get();
    // 判断发送的消息是否得到broker的确认
    boolean confirmAck = confirm.isAck();
    if (confirmAck){
        // 发送延迟等待消息
        rabbitTemplate.convertAndSend("x.ttl","k.ttl" , order.getOrderId());
    }

### 四、总结 ###

到这里，基本就实现了整个订单延迟自动取消的思路，但事实上还有问题。

投递订单消息到MQ后要投递多一份到延迟队列，可能存在第一次投递成功但投递到延迟队列失败的情况，这里则需要依赖分布式锁或者增加补偿机制；还有编码上的问题，MQ队列名称这些最好抽离出来，当然这里只是demo，就没有那么规范，如果是产品开发，这些都需要最好规定，方便后期维护。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hzZjE1NzY4NjE1Mjg0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221021/3df47bc292724794adf6f197a51c5b53.png