当前位置：首页 > news >正文

SpringBoot RabbitMQ 延时队列取消订单【SpringBoot系列14】

news 2025/8/21 4:14:54

SpringCloud 大型系列课程正在制作中，欢迎大家关注与提意见。
程序员每天的CV 与板砖，也要知其所以然，本系列课程可以帮助初学者学习 SpringBooot 项目开发与 SpringCloud 微服务系列项目开发

1 项目准备

SpringBoot 雪花算法生成商品订单号【SpringBoot系列13】本文章基于这个项目来开发

本文章是系列文章，每节文章都有对应的代码，每节的源码都是在上一节的基础上配置而来，对应的视频讲解课程正在火速录制中。

订单系统，用户下单，即要保存即时性，也要保证流畅性，同时还要防止超卖，本文章是基于 RabbitMQ 消息队列 + Redis 实现的下单，当然后续还会的秒杀系统设计以及后续的微服务以及熔断控制等等

如下图所示是本项目实现的一个下单流程的主要过程：

在这里插入图片描述

本文章实现的是当用户下单成功后10分钟内没有支付时，使用 RabbitMQ 延时消息队列取消订单，并恢复商品的库存。

1 RabbitMQ 延时插件的加载

插件地址：

https://www.rabbitmq.com/community-plugins.html

在这里插入图片描述
下载对应的版本到电脑本地，打开终端将下载的压缩包移动到plugins目录下

docker cp /Users/androidlongs/Downloads/rabbitmq_delayed_message_exchange-3.9.0.ez  rabbitmq:/plugins

然后进入容器，我这里使用容器名字也可以用容器id进入

docker exec -it rabbitmq /bin/bash

移动到plugins目录下

cd plugins

查看是否上传成功

ls

也可以使用

rabbitmq-plugins list

然后启用延时插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2 SpringBoot RabbitMQ 延时消息

2.1 创建消息队列

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/*** 插件配置类*/
@Configuration
public class TestDelayedMessageConfig {public static final String DIRECT_QUEUE = "test.queue.direct";//队列public static final String DELAYED_EXCHANGE = "test.exchange.delayed";//延迟交换机public static final String ROUTING_KEY = "test.routingkey.bind";//绑定的routing-key/*** 定义队列**/@Beanpublic Queue directQueue(){return new Queue(DIRECT_QUEUE,true);}/*** 定义延迟交换机* args:根据该参数进行灵活路由，设置为“direct”，意味着该插件具有与直连交换机具有相同的路由行为* 交换机类型为 x-delayed-message**/@Beanpublic CustomExchange delayedExchange(){Map<String, Object> args = new HashMap<String, Object>();args.put("x-delayed-type", "direct");return new CustomExchange(DELAYED_EXCHANGE, "x-delayed-message", true, false, args);}/*** 队列和延迟交换机绑定**/@Beanpublic Binding orderBinding() {return BindingBuilder.bind(directQueue()).to(delayedExchange()).with(ROUTING_KEY).noargs();}}

CustomExchange 是自定义交换机，一般是配合插件来使用的。

2.2 创建延时消息生产者

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service
@Slf4j
public class TestDelayedMQSender {@Autowiredprivate RabbitTemplate rabbitTemplate;public void testSsend(String msg, Integer delayTime) {log.info("测试发送延时消息 {} : {}", delayTime, msg);//将消息携带路由键值rabbitTemplate.convertAndSend(TestDelayedMessageConfig.DELAYED_EXCHANGE,//交换机名称TestDelayedMessageConfig.ROUTING_KEY,//路由 key msg,message -> {//设置延时的时间  单位毫秒message.getMessageProperties().setDelay(delayTime);return message;});}}

2.3 创建延时消息消费者

import lombok.extern.log4j.Log4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@Log4j
@RabbitListener(queues = TestDelayedMessageConfig.DIRECT_QUEUE)//监听队列名称
public class TestDelayedMQReciever {@RabbitHandlerpublic void process(String message){log.info("DelayedMQReciever接收到的消息是："+ message);}
}

2.4 测试

@Api(tags = "订单测试模块")
@RestController()
@RequestMapping("/test/orders")
@Slf4j
public class OrderTestController {@AutowiredTestDelayedMQSender delayedMQSender;@GetMapping("/send/delay")public R createPreOrder(@RequestParam(required = false,defaultValue = "10000")  Integer delayTime) {delayedMQSender.testSsend("测试延时消息了",delayTime);return R.ok();}}