使用RabbitMQ处理超时订单

下载插件

基于死信队列虽然可以实现延迟消息完成超时订单取消，但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果。

插件下载地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange

因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。

docker volume inspect mq-plugins

结果如下：

[
    {
        "CreatedAt": "2024-06-19T09:22:59+08:00",
        "Driver": "local",
        "Labels": null,
        "Mountpoint": "/var/lib/docker/volumes/mq-plugins/_data",
        "Name": "mq-plugins",
        "Options": null,
        "Scope": "local"
    }
]

插件目录被挂载到了/var/lib/docker/volumes/mq-plugins/_data这个目录，我们上传插件到该目录下。

接下来执行命令，安装插件：

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

定义常量

无论是消息发送还是接收都是在交易服务完成，因此我们在constants中定义一个常量类，用于记录交换机、队列、RoutingKey等常量：

内容如下：

package com.hmall.trade.constants;

public interface MQConstants {
    String DELAY_EXCHANGE_NAME = "trade.delay.direct";
    String DELAY_ORDER_QUEUE_NAME = "trade.delay.order.queue";
    String DELAY_ORDER_KEY = "delay.order.query";
}

配置MQ

1）添加依赖：

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2）配置MQ地址：

spring:
  rabbitmq:
    host: 192.168.10.130 # 你的虚拟机IP
    port: 5672 # 端口
    virtual-host: /skytakeout # 虚拟主机
    username: cyan # 用户名
    password: 123456 # 密码

在Virtual Hosts 中创建一个项目专用的virtual host：/skytakeout

一般是配置为多环境模式

application.yaml

spring:
  rabbitmq:
    host: ${mq.host} # 你的虚拟机IP
    port: 5672 # 端口
    virtual-host: ${mq.virtual-host} # 虚拟主机
    username: ${mq.username} # 用户名
    password: ${mq.password} # 密码

application-dev.yaml

mq:
  host: 192.168.10.130 # 你的虚拟机IP
  virtual-host: /skytakeout # 虚拟主机
  username: cyan # 用户名
  password: 123456 # 密码

application-prod.yaml

mq:
  host: mq # docker网络别名
  virtual-host: /skytakeout # 虚拟主机
  username: cyan # 用户名
  password: 123456 # 密码

发送消息

//5,发送延迟消息，检测订单支付状态
rabbitTemplate.convertAndSend(
	MQConstants.DELAY_EXCHANGE_NAME,
    MQConstants.DELAY_ORDER_KEY,
    order.getId(),
    message ->{
    	message.getMessageProperties().setDelay(10000);
    	return message;
	});

监听消息

package com.sky.listener;

import com.sky.constant.MQConstant;
import com.sky.entity.Orders;
import com.sky.mapper.OrderMapper;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

@Component
@RequiredArgsConstructor
public class OrderDelayMessageListener {

    private final OrderMapper orderMapper;
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = MQConstant.DELAY_ORDER_QUEUE_NAME),
            exchange = @Exchange(name = MQConstant.DELAY_EXCHANGE_NAME, delayed = "true"),
            key = MQConstant.DELAY_ORDER_KEY
    ))
    public void listenOrderDelayMessage(Long orderId) {
        // 根据id查询订单
        Orders order = orderMapper.getById(orderId);

        // 2.检测订单状态，判断是否已支付
        if(order == null || order.getStatus() != Orders.PENDING_PAYMENT){
            // 订单不存在或者已经支付
            return;
        }

        order.setStatus(Orders.CANCELLED);
        order.setCancelReason("支付超时，自动取消");
        order.setCancelTime(LocalDateTime.now());
        orderMapper.update(order);
    }

}