RabbitMQ
RabbitMQ是一个被广泛使用的开源消息队列。它是轻量级且易于部署的,它能支持多种消息协议。RabbitMQ可以部署在分布式和联合配置中,以满足高规模、高可用性的需求。
RabbitMQ的安装和使用
安装Erlang
下载地址:http://erlang.org/download/otp_win64_21.3.exe
安装RabbitMQ
下载地址:https://dl.bintray.com/rabbitmq/all/rabbitmq-server/3.7.14/rabbitmq-server-3.7.14.exe
安装完成后,进入RabbitMQ安装目录下的sbin目录
在地址栏输入cmd并回车启动命令行,然后输入以下命令启动管理功能:
rabbitmq-plugins enable rabbitmq_management
访问地址查看是否安装成功:http://localhost:15672/
输入账号密码并登录:guest guest
创建帐号并设置其角色为管理员:mall mall
创建一个新的虚拟host为:/mall
点击mall用户进入用户配置页面
给mall用户配置该虚拟host的权限
RabbitMQ的消息模型
整合RabbitMQ实现延迟消息
模拟应用场景:用户下单以后,订单超时如何取消订单的问题
用户进行下单操作(会有锁定商品库存、使用优惠券、积分一系列的操作);
1.生成订单,获取订单的id;
2.获取到设置的订单超时时间(假设设置的为60分钟不支付取消订单);
3.按订单超时时间发送一个延迟消息给RabbitMQ,让它在订单超时后触发取消订单的操作;
4.如果用户没有支付,进行取消订单操作
在pom.xml中添加相关依赖
<!--消息队列相关依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!--lombok依赖--><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency>
修改SpringBoot配置文件
修改application.yml文件,在spring节点下添加Mongodb相关配置
rabbitmq:
host: localhost # rabbitmq的连接地址
port:5672 # rabbitmq的连接端口号
virtual-host:/mall # rabbitmq的虚拟host
username: mall # rabbitmq的用户名
password: mall # rabbitmq的密码
publisher-confirms:true #如果对异步消息需要回调必须设置为true
添加消息队列的枚举配置类QueueEnum
用于延迟消息队列及处理取消订单消息队列的常量定义,包括交换机名称、队列名称、路由键名称
packageorg.yxin.entity;importlombok.Getter;/**
* 消息队列枚举配置
* Created by dyx on 2024/2/16.
*/@GetterpublicenumQueueEnum{/**
* 消息通知队列
*/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"),/**
* helloworld队列
*/QUEUE_HELLO_WORLD("mall.hello.world","mall.hello.world","mall.hello.world");/**
* 交换名称
*/privateString exchange;/**
* 队列名称
*/privateString name;/**
* 路由键
*/privateString routeKey;QueueEnum(String exchange,String name,String routeKey){this.exchange = exchange;this.name = name;this.routeKey = routeKey;}}
添加RabbitMQ的配置
packageorg.yxin.config;importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.yxin.entity.QueueEnum;/**
1. 消息队列配置
2. Created by dyx on 2024/2/16.
*/@ConfigurationpublicclassRabbitMqConfig{/**
* 订单消息实际消费队列所绑定的交换机
*/@BeanDirectExchangeorderDirect(){return(DirectExchange)ExchangeBuilder.directExchange(QueueEnum.QUEUE_ORDER_CANCEL.getExchange()).durable(true).build();}/**
* 订单延迟队列所绑定的交换机
*/@BeanDirectExchangeorderTtlDirect(){return(DirectExchange)ExchangeBuilder.directExchange(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange()).durable(true).build();}/**
* helloworld交换机
*/@BeanDirectExchangehelloDirect(){return(DirectExchange)ExchangeBuilder.directExchange(QueueEnum.QUEUE_HELLO_WORLD.getExchange()).durable(true).build();}/**
* 订单实际消费队列
*/@BeanpublicQueueorderQueue(){returnnewQueue(QueueEnum.QUEUE_ORDER_CANCEL.getName());}/**
* 订单延迟队列(死信队列)
*/@BeanpublicQueueorderTtlQueue(){returnQueueBuilder.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();}/**
* helloworld队列
*/@BeanpublicQueuehelloQueue(){returnnewQueue(QueueEnum.QUEUE_HELLO_WORLD.getName());}/**
* 将订单队列绑定到交换机
*/@BeanBindingorderBinding(DirectExchange orderDirect,Queue orderQueue){returnBindingBuilder.bind(orderQueue).to(orderDirect).with(QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey());}/**
* 将订单延迟队列绑定到交换机
*/@BeanBindingorderTtlBinding(DirectExchange orderTtlDirect,Queue orderTtlQueue){returnBindingBuilder.bind(orderTtlQueue).to(orderTtlDirect).with(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey());}/**
* 将hello队列绑定到交换机
*/@BeanBindinghelloBinding(DirectExchange helloDirect,Queue helloQueue){returnBindingBuilder.bind(helloQueue).to(helloDirect).with(QueueEnum.QUEUE_HELLO_WORLD.getRouteKey());}}
在RabbitMQ管理页面可以看到以下交换机和队列
交换机及队列说明
- mall.order.direct(取消订单消息队列所绑定的交换机):绑定的队列为mall.order.cancel,一旦有消息以mall.order.cancel为路由键发过来,会发送到此队列。 mall.order.direct.ttl(订单延迟消息队列所绑定的交换机):绑定的队列为
- mall.order.cancel.ttl,一旦有消息以mall.order.cancel.ttl为路由键发送过来,会转发到此队列,并在此队列保存一定时间,等到超时后会自动将消息发送到mall.order.cancel(取消订单消息消费队列)
添加延迟消息的发送者CancelOrderSender
packageorg.yxin.component;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.amqp.AmqpException;importorg.springframework.amqp.core.AmqpTemplate;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.core.MessagePostProcessor;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;importorg.yxin.entity.QueueEnum;/**
* 取消订单消息的发出者
* Created by dyx on 2024/2/16.
*/@ComponentpublicclassCancelOrderSender{privatestaticLoggerLOGGER=LoggerFactory.getLogger(CancelOrderSender.class);@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendMessage(Long orderId,finallong delayTimes){//给延迟队列发送消息
amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(),QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), orderId,newMessagePostProcessor(){@OverridepublicMessagepostProcessMessage(Message message)throwsAmqpException{//给消息设置延迟毫秒值
message.getMessageProperties().setExpiration(String.valueOf(delayTimes));return message;}});LOGGER.info("send delay message orderId:{}",orderId);}}
添加取消订单消息的接收者CancelOrderReceiver
用于从取消订单的消息队列(mall.order.cancel)里接收消息
packageorg.yxin.component;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;importorg.yxin.service.OmsPortalOrderService;/**
* 取消订单消息的处理者
* Created by dyx on 2024/2/16.
*/@Component@RabbitListener(queues ="mall.order.cancel")publicclassCancelOrderReceiver{privatestaticLoggerLOGGER=LoggerFactory.getLogger(CancelOrderReceiver.class);@AutowiredprivateOmsPortalOrderService portalOrderService;@RabbitHandlerpublicvoidhandle(Long orderId){LOGGER.info("receive delay message orderId:{}",orderId);
portalOrderService.cancelOrder(orderId);}}
添加OmsPortalOrderService接口
packageorg.yxin.service;importorg.springframework.transaction.annotation.Transactional;importorg.yxin.common.api.CommonResult;importorg.yxin.entity.OrderParam;/**
* 前台订单管理Service
* Created by dyx on 2024/2/16.
*/publicinterfaceOmsPortalOrderService{/**
* 根据提交信息生成订单
*/@TransactionalCommonResultgenerateOrder(OrderParam orderParam);/**
* 取消单个超时订单
*/@TransactionalvoidcancelOrder(Long orderId);}
添加OmsPortalOrderService的实现类OmsPortalOrderServiceImpl
packageorg.yxin.service.impl;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;importorg.yxin.common.api.CommonResult;importorg.yxin.component.CancelOrderSender;importorg.yxin.component.HelloOrderSender;importorg.yxin.entity.OrderParam;importorg.yxin.service.OmsPortalOrderService;/**
* 前台订单管理Service
* Created by dyx on 2023/8/30.
*/@ServicepublicclassOmsPortalOrderServiceImplimplementsOmsPortalOrderService{privatestaticLoggerLOGGER=LoggerFactory.getLogger(OmsPortalOrderServiceImpl.class);@AutowiredprivateCancelOrderSender cancelOrderSender;@AutowiredprivateHelloOrderSender helloOrderSender;@OverridepublicCommonResultgenerateOrder(OrderParam orderParam){//todo 执行一系类下单操作,具体参考mall项目LOGGER.info("process generateOrder");//下单完成后开启一个延迟消息,用于当用户没有付款时取消订单(orderId应该在下单后生成)sendDelayMessageCancelOrder(11L);returnCommonResult.success(null,"下单成功");}@OverridepublicvoidcancelOrder(Long orderId){LOGGER.info("process cancelOrder orderId:{}",orderId);}privatevoidsendDelayMessageCancelOrder(Long orderId){//获取订单超时时间,假设为60分钟(测试用的30秒)long delayTimes =30*1000;//发送延迟消息
cancelOrderSender.sendMessage(orderId, delayTimes);
helloOrderSender.sendMessage(orderId, delayTimes);}}
添加OmsPortalOrderController定义接口
packageorg.yxin.controller;importio.swagger.annotations.Api;importio.swagger.annotations.ApiOperation;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Controller;importorg.springframework.web.bind.annotation.RequestBody;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RequestMethod;importorg.springframework.web.bind.annotation.ResponseBody;importorg.yxin.entity.OrderParam;importorg.yxin.service.OmsPortalOrderService;/**
* 订单管理Controller
* Created by dyx on 2023/8/30.
*/@Controller@Api(tags ="OmsPortalOrderController", description ="订单管理")@RequestMapping("/order")publicclassOmsPortalOrderController{@AutowiredprivateOmsPortalOrderService portalOrderService;@ApiOperation("根据购物车信息生成订单")@RequestMapping(value ="/generateOrder", method =RequestMethod.POST)@ResponseBodypublicObjectgenerateOrder(@RequestBodyOrderParam orderParam){return portalOrderService.generateOrder(orderParam);}}
进行接口测试
调用下单接口
注意:已经将延迟消息时间设置为30秒
版权归原作者 dyx1472581 所有, 如有侵权,请联系我们删除。