Mall脚手架总结(四) —— SpringBoot整合RabbitMQ实现超时订单处理

前言

        在电商项目中，订单因为某种特殊情况被取消或者超时未支付都是比较常规的用户行为，而实现该功能我们就要借助消息中间件来为我们维护这么一个消息队列。在mall脚手架中选择了RabbitMQ消息中间件，接下来荔枝就会根据功能需求来梳理一下超时订单处理功能以及相应的背景知识。希望对正在学习的小伙伴有帮助~~~

文章目录

前言

一、整合RabbitMQ实现延时消息

1.1 RabbitMQ管理界面的VirtualHost

1.2 回顾：枚举类的优点

1.3 划重点：Spring AMQP框架 

1.3.1 AMQPTemplate

1.3.2 Message

1.3.3 @RabbitListener注解

1.4 订单超时未支付取消订单的流程

总结

一、整合RabbitMQ实现延时消息

1.1 RabbitMQ管理界面的VirtualHost

        Virtual Host虚拟主机，相当于是一个个的相对独立的RabbitMQ服务器。每个虚拟主机都有自己独立的用户、权限、交换机（exchange）、队列（queue）和绑定关系。在RabbitMQ中，每个连接到服务器的客户端都必须选择一个虚拟主机进行操作。如果客户端没有指定虚拟主机，默认会使用/作为虚拟主机，也就是RabbitMQ默认的虚拟主机。为保证隔离性，这里声明了一个/mall的虚拟主机，与我们默认的用户虚拟主机/隔离开来。

1.2 回顾：枚举类的优点

枚举类Enum是一种特殊的数据库类型，用于表示固定数量的命名常量，枚举类定义了一个新的数据类型，该类型可以包含一组预定义的值。

优点：

• 安全性：保证方法参数的类型的安全性，避免非法值的传入；
• 避免魔法数字： 枚举类可以帮助你避免在代码中使用魔法数字（magic numbers）。魔法数字是指在代码中直接使用的、没有明确含义的数字。使用枚举类可以为这些常量提供有意义的名字。
• 支持迭代器（Iteration）： 枚举类提供了一种便捷的方式来遍历枚举值。

所以在脚手架中我们定义交换机名称、队列名称和routingKey就借助了枚举类。

/**
 * @auther lzddl
 * @description 消息队列枚举配置
 */
@Getter
public enum QueueEnum {
    /**
     * 消息通知队列
     */
    QUEUE_ORDER_CANCEL("mall.order.direct", "mall.order.cancel", "mall.order.cancel"),
    /**
     * 消息通知ttl队列
     */
    QUEUE_TTL_ORDER_CANCEL("mall.order.direct.ttl", "mall.order.cancel.ttl", "mall.order.cancel.ttl");
 
    /**
     * 交换机名称
     */
    private String exchange;
    /**
     * 队列名称
     */
    private String name;
    /**
     * 路由键
     */
    private String routeKey;
 
    QueueEnum(String exchange, String name, String routeKey) {
        this.exchange = exchange;
        this.name = name;
        this.routeKey = routeKey;
    }
}

1.3 划重点：Spring AMQP框架 

        Spring AMQP是一个基于AMQP协议的强大的消息中间件框架，它提供了一个简单的API来发送和接收异步、可靠的消息。AMQP是Spring框架的一部分，可以与Spring Boot和其他Spring项目一起使用。Spring AMQP支持多种消息协议，包括RabbitMQ、Apache ActiveMQ和Qpid等。它提供了一个高级的消息模型，包括消息确认、事务和消息监听器等功能，使得开发者可以轻松地编写可靠的消息应用程序。同时Spring AMQP还提供了一些高级特性，如消息转换器、消息路由、消息过滤和消息拦截等。总的来说，Spring AMQP 是对 Spring 基于 AMQP 的消息收发解决方案，在SpringBoot项目中操作消息中间件RabbitMQ的相关操作的时候，我们需要借助Spring提供的AMQP框架！

AMQP：高级消息队列协议，是面向消息的中间件的平台中立的线级协议。

Spring AMQP的核心组件：

• ConnectionFactory：连接工厂接口，用于创建连接。

• AmqpAdmin：封装了RabbitMQ的基础管理操作，比如对交换机、队列、绑定的声明和删除等。
• Message：Spring AMQP 对消息的封装。两个重要的属性：body：消息内容；messageProperties：消息属性。

• 消息模板AmqpTemplate： 用来简化消息的收发，支持消息的确认与返回。跟 JDBCTemplate一 样，它封装了创建连接 、创建消息信道、收发消息、消息格式转换、关闭信道、关闭连接等等操作。

• 消息监听Messager Listener： Spring AMQP 异步消息投递的监听器接口，它只有一个方法onMessage，用于处理消息队列推送来的消息。

• 转换器MessageConvertor：用来处理消息对象的序列化和反序列化的操作工具，它可以将消息对象转换为消息队列可以处理的格式，并将接收到的消息转换为Java对象。

1.3.1 AMQPTemplate

AMQPTemplate是Spring AMQP框架提供的一个接口，它定义了一系列用于发送和接收消息的方法。我们来看看源码并归类一下这些方法：

public interface AmqpTemplate {
    void send(Message var1) throws AmqpException;
    void send(String var1, Message var2) throws AmqpException;
    void send(String var1, String var2, Message var3) throws AmqpException;
 
 
    void convertAndSend(Object var1) throws AmqpException;
    void convertAndSend(String var1, Object var2) throws AmqpException;
    void convertAndSend(String var1, String var2, Object var3) throws AmqpException;
    void convertAndSend(Object var1, MessagePostProcessor var2) throws AmqpException;
    void convertAndSend(String var1, Object var2, MessagePostProcessor var3) throws AmqpException;
    void convertAndSend(String var1, String var2, Object var3, MessagePostProcessor var4) throws AmqpException;
 
 
    @Nullable
    Message receive() throws AmqpException;
    @Nullable
    Message receive(String var1) throws AmqpException;
    @Nullable
    Message receive(long var1) throws AmqpException;
    @Nullable
    Message receive(String var1, long var2) throws AmqpException;
 
 
    @Nullable
    Object receiveAndConvert() throws AmqpException;
    @Nullable
    Object receiveAndConvert(String var1) throws AmqpException;
    @Nullable
    Object receiveAndConvert(long var1) throws AmqpException;
    @Nullable
    Object receiveAndConvert(String var1, long var2) throws AmqpException;
    @Nullable
     T receiveAndConvert(ParameterizedTypeReference var1) throws AmqpException;
    @Nullable
     T receiveAndConvert(String var1, ParameterizedTypeReference var2) throws AmqpException;
    @Nullable
     T receiveAndConvert(long var1, ParameterizedTypeReference var3) throws AmqpException;
    @Nullable
     T receiveAndConvert(String var1, long var2, ParameterizedTypeReference var4) throws AmqpException;
 
 
     boolean receiveAndReply(ReceiveAndReplyCallback var1) throws AmqpException;
     boolean receiveAndReply(String var1, ReceiveAndReplyCallback var2) throws AmqpException;
     boolean receiveAndReply(ReceiveAndReplyCallback var1, String var2, String var3) throws AmqpException;
     boolean receiveAndReply(String var1, ReceiveAndReplyCallback var2, String var3, String var4) throws AmqpException;
     boolean receiveAndReply(ReceiveAndReplyCallback var1, ReplyToAddressCallback var2) throws AmqpException;
     boolean receiveAndReply(String var1, ReceiveAndReplyCallback var2, ReplyToAddressCallback var3) throws AmqpException;
 
 
    @Nullable
    Message sendAndReceive(Message var1) throws AmqpException;
    @Nullable
    Message sendAndReceive(String var1, Message var2) throws AmqpException;
    @Nullable
    Message sendAndReceive(String var1, String var2, Message var3) throws AmqpException;
   
 
    @Nullable
    Object convertSendAndReceive(Object var1) throws AmqpException;
    @Nullable
    Object convertSendAndReceive(String var1, Object var2) throws AmqpException;
    @Nullable
    Object convertSendAndReceive(String var1, String var2, Object var3) throws AmqpException;
    @Nullable
    Object convertSendAndReceive(Object var1, MessagePostProcessor var2) throws AmqpException;
    @Nullable
    Object convertSendAndReceive(String var1, Object var2, MessagePostProcessor var3) throws AmqpException;
    @Nullable
    Object convertSendAndReceive(String var1, String var2, Object var3, MessagePostProcessor var4) throws AmqpException;
 
 
    @Nullable
     T convertSendAndReceiveAsType(Object var1, ParameterizedTypeReference var2) throws AmqpException;
    @Nullable
     T convertSendAndReceiveAsType(String var1, Object var2, ParameterizedTypeReference var3) throws AmqpException;
    @Nullable
     T convertSendAndReceiveAsType(String var1, String var2, Object var3, ParameterizedTypeReference var4) throws AmqpException;
    @Nullable
     T convertSendAndReceiveAsType(Object var1, MessagePostProcessor var2, ParameterizedTypeReference var3) throws AmqpException;
    @Nullable
     T convertSendAndReceiveAsType(String var1, Object var2, MessagePostProcessor var3, ParameterizedTypeReference var4) throws AmqpException;
    @Nullable
     T convertSendAndReceiveAsType(String var1, String var2, Object var3, MessagePostProcessor var4, ParameterizedTypeReference var5) throws AmqpException;
}

 //给延迟队列发送消息
amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), orderId, new MessagePostProcessor() {
    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        //给消息设置延迟毫秒值
        message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
        return message;
    }
});

message.getMessageProperties().setExpiration(String.valueOf(delayTimes));

@Component
@RabbitListener(queues = "mall.order.cancel")
public class CancelOrderReceiver {
    private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderReceiver.class);
    @Autowired
    private OmsPortalOrderService portalOrderService;
    @RabbitHandler
    public void handle(Long orderId){
        LOGGER.info("receive delay message orderId:{}",orderId);
        portalOrderService.cancelOrder(orderId);
    }
}

    /**
     * 订单延迟队列（死信队列）
     */
    @Bean
    public Queue orderTtlQueue() {
        return QueueBuilder
                .durable(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getName())
                .withArgument("x-dead-letter-exchange", QueueEnum.QUEUE_ORDER_CANCEL.getExchange())//到期后转发的交换机
                .withArgument("x-dead-letter-routing-key", QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey())//到期后转发的路由键
                .build();
    }