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> T receiveAndConvert(ParameterizedTypeReference<T> var1) throws AmqpException;
    @Nullable
    <T> T receiveAndConvert(String var1, ParameterizedTypeReference<T> var2) throws AmqpException;
    @Nullable
    <T> T receiveAndConvert(long var1, ParameterizedTypeReference<T> var3) throws AmqpException;
    @Nullable
    <T> T receiveAndConvert(String var1, long var2, ParameterizedTypeReference<T> var4) throws AmqpException;


    <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> var1) throws AmqpException;
    <R, S> boolean receiveAndReply(String var1, ReceiveAndReplyCallback<R, S> var2) throws AmqpException;
    <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> var1, String var2, String var3) throws AmqpException;
    <R, S> boolean receiveAndReply(String var1, ReceiveAndReplyCallback<R, S> var2, String var3, String var4) throws AmqpException;
    <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> var1, ReplyToAddressCallback<S> var2) throws AmqpException;
    <R, S> boolean receiveAndReply(String var1, ReceiveAndReplyCallback<R, S> var2, ReplyToAddressCallback<S> 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> T convertSendAndReceiveAsType(Object var1, ParameterizedTypeReference<T> var2) throws AmqpException;
    @Nullable
    <T> T convertSendAndReceiveAsType(String var1, Object var2, ParameterizedTypeReference<T> var3) throws AmqpException;
    @Nullable
    <T> T convertSendAndReceiveAsType(String var1, String var2, Object var3, ParameterizedTypeReference<T> var4) throws AmqpException;
    @Nullable
    <T> T convertSendAndReceiveAsType(Object var1, MessagePostProcessor var2, ParameterizedTypeReference<T> var3) throws AmqpException;
    @Nullable
    <T> T convertSendAndReceiveAsType(String var1, Object var2, MessagePostProcessor var3, ParameterizedTypeReference<T> var4) throws AmqpException;
    @Nullable
    <T> T convertSendAndReceiveAsType(String var1, String var2, Object var3, MessagePostProcessor var4, ParameterizedTypeReference<T> var5) throws AmqpException;
}

从源码中我们可以了解到，其实该接口提供了八种类型的消息操作方法，因为不同方法都采用了重载所以看起来有点吓人，接下来我们大致根据方法名弄清楚这些方法的功能即可。

send()和receive()就不说了，根据给出的参数来发送消息和接收消息

convertAndSend：转化并发送消息；
receiveAndConvert：接收并转化消息
receiveAndReply：接收并回复消息
sendAndReceive：发送并接收消息
convertSendAndReceive：转化发送和接收
convertSendAndReceiveAsType：根据类型来发送接收

其实AMQPTemplate是一个比较抽象的接口，其中操作RabbitMQ更为具体的接口的实现类是RabbitMQTemplate。而关于RabbitMQTemplate的源码这里就不展示了（一千多行呢~）

 //给延迟队列发送消息
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;
    }
});

可以看到我们调用了RabbitMQTemplate实现的AMQPTemplate接口方法来发送消息。

1.3.2 Message

Message消息，是服务器与应用程序之间传递的数据，由Properties和Body组成， Properties可以对消息进行修饰，如消息的优先级、传输格式（如JSON）、延迟等高级特性，Body则就是消息体内容。

Message类的使用场景：

发送消息时：通常使用最多的是编写消息体内容、设置过期时间、设置持久化；发送消息的类型分为两种情况：基本类型和对象；
接收消息时：实际是把二进制byte转为需要的类型，再进行数据传递和业务处理，接收消息的类型分为两种情况：基本类型和对象

在脚手架中我们看到使用了Message来反取消息对象的属性并设置相应的消息过期时间：

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

其余的核心组件具体内容可以参考这位大佬的博文： https://blog.csdn.net/weixin_45596022/article/details/113359009

1.3.3 @RabbitListener注解

@RabbitListener注解是Spring AMQP框架提供的注解，用于简化RabbitMQ消息消费者的创建。当你在方法上使用@RabbitListener注解时，Spring会自动创建一个RabbitMQ消息监听器，用于监听指定队列上的消息，并在消息到达时调用被注解的方法来处理消息。

因此在脚手架中我们通过该注解来实现死信队列消费者的创建：

@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);
    }
}

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

在mall脚手架中模拟了一个订单到期未支付取消下单的操作，首先用户在购买之后会创建订单（随之可能会有锁定库存、判断会员身份或者积分、优惠券等操作），Controller层中的generateOrder()就会创建一个带有过期时间的延时消息，这部分是通过一个之前已经定义好了的CancelOrderSender也就是延时消息的发送者类发送延时消息到死信队列中。这里我们来看一下配置死信队列的配置类方法：

    /**
     * 订单延迟队列（死信队列）
     */
    @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();
    }

由于我们定义了一个死信队列和一个延时队列，订单下单之后我们就会把消息发送到死信队列，由于用户迟迟不支付，所以死信队列中的消息一直没有被消费，等到TTL时间一到就会转发到普通队列中被消费者消费。前面在讲@RabbitListener注解的时候已经给出了消费者的demo，消费者监听的就是普通队列。当消费者消费后会触发取消订单的API进行订单取消的操作（释放库存、扣除优惠券或积分等操作）