一、什么是RabbitMQ死信队列
RabbitMQ死信队列(Dead-Letter Exchange,简称DLX)是一种特殊类型的交换机,用于处理在队列中无法被消费的消息。当消息无法被消费时,它会被转发到死信队列中,以便进一步处理。
在RabbitMQ中,死信队列通常用于处理以下情况:
- 消息无法被消费者处理:例如,如果消费者崩溃或消息的格式不正确,则无法处理消息。此时,消息将被发送到死信队列进行进一步处理。
- 消息的优先级较低:如果消息的优先级较低,则可能无法在队列中得到及时处理。在这种情况下,消息也会被发送到死信队列中,以确保它最终被处理。
要使用死信队列,需要创建一个普通的交换机和一个普通的队列,然后创建一个死信队列并将其绑定到普通队列上。当消息无法被消费时,它将被发送到死信队列中。
二、RabbitMQ关单逻辑
1. 流程图
- 订单创建成功后, 发送消息给order-event-exchange交换机,采用路由键order.create.order
- order-event-exchange交换机将消息转发给order.delay.queue队列,队列保存时间为30分钟,如果没有消费,则再将消息路由到order-event-exchange交换机,采用路由键order.release.order
- order-event-exchange交换机再将消息转发到死信队列 order-realease-order.queue,采用路由键order.release.order
- 监听死信队列 order-realease-order.queue,如果订单状态为“待付款”的,说明支付不成功,改为“取消”关闭订单
三、Springboot配置RabbitMQ
1. 添加依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2. 参数配置
spring:rabbitmq:host: 127.0.0.1port: 5672# 虚拟主机virtual-host: /# 开启发送端发送确认,无论是否到达broker都会触发回调【发送端确认机制+本地事务表】publisher-confirm-type: correlated# 开启发送端抵达队列确认,消息未被队列接收时触发回调【发送端确认机制+本地事务表】publisher-returns: true# 消息在没有被队列接收时是否强行退回template:mandatory: true# 消费者手动确认模式,关闭自动确认,否则会消息丢失listener:simple:acknowledge-mode: manual
3、RabbitMQ模板配置
@Configuration
@Slf4j
public class MyRabbitConfig {@Primary@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {// TODO 封装RabbitTemplateRabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMessageConverter(messageConverter());initRabbitTemplate(rabbitTemplate);return rabbitTemplate;}@Beanpublic MessageConverter messageConverter() {// 使用json序列化器来序列化消息,发送消息时,消息对象会被序列化成json格式return new Jackson2JsonMessageConverter();}/*** 定制RabbitTemplate* 1、服务收到消息就会回调* 1、spring.rabbitmq.publisher-confirms: true* 2、设置确认回调* 2、消息正确抵达队列就会进行回调* 1、spring.rabbitmq.publisher-returns: true* spring.rabbitmq.template.mandatory: true* 2、设置确认回调ReturnCallback* <p>* 3、消费端确认(保证每个消息都被正确消费,此时才可以broker删除这个消息)*///@PostConstruct // (MyRabbitConfig对象创建完成以后,执行这个方法)public void initRabbitTemplate(RabbitTemplate rabbitTemplate) {/*** 发送消息触发confirmCallback回调* @param correlationData:当前消息的唯一关联数据(如果发送消息时未指定此值,则回调时返回null)* @param ack:消息是否成功收到(ack=true,消息抵达Broker)* @param cause:失败的原因*/rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {log.info("发送消息触发confirmCallback回调" +"\ncorrelationData ===> " + correlationData +"\nack ===> " + ack + "" +"\ncause ===> " + cause);log.info("=================================================");});/*** 消息未到达队列触发returnCallback回调* 只要消息没有投递给指定的队列,就触发这个失败回调* @param message:投递失败的消息详细信息* @param replyCode:回复的状态码* @param replyText:回复的文本内容* @param exchange:接收消息的交换机* @param routingKey:接收消息的路由键*/rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {log.info("消息未到达队列触发returnCallback回调" +"\nmessage ===> " + message +"\nreplyCode ===> " + replyCode +"\nreplyText ===> " + replyText +"\nexchange ===> " + exchange +"\nroutingKey ===> " + routingKey);// TODO 修改mq_message,设置消息状态为2-错误抵达【后期定时器重发消息】});}
}
4、启动RabbitMQ
@SpringBootApplication
@EnableRabbit //启用RabbitMQ自动配置
public class Application implements CommandLineRunner { public static void main(String[] args) throws Exception {SpringApplication.run(Application.class, args);}
}
四、关单业务流程
1. 提交订单完成后,发送关单消息
1. 提交订单控制层
/*** 创建订单* 创建成功,跳转订单支付页* 创建失败,跳转结算页* 无需提交要购买的商品,提交订单时会实时查询最新的购物车商品选中数据提交*/@TokenVerify@PostMapping(value = "/submitOrder")public String submitOrder(OrderSubmitVO vo, Model model, RedirectAttributes attributes) {try {SubmitOrderResponseVO orderVO = orderService.submitOrder(vo);// 创建订单成功,跳转收银台model.addAttribute("submitOrderResp", orderVO);// 封装VO订单数据,供页面解析[订单号、应付金额]return "pay";} catch (Exception e) {// 下单失败回到订单结算页if (e instanceof VerifyPriceException) {String message = ((VerifyPriceException) e).getMessage();attributes.addFlashAttribute("msg", "下单失败" + message);} else if (e instanceof NoStockException) {String message = ((NoStockException) e).getMessage();attributes.addFlashAttribute("msg", "下单失败" + message);}return "redirect:http://order.kmall.com/toTrade";}}
2. 提交订单实现层
@Transactional(isolation = Isolation.DEFAULT)@Overridepublic SubmitOrderResponseVO submitOrder(OrderSubmitVO orderSubmitVO) throws Exception {SubmitOrderResponseVO result = new SubmitOrderResponseVO();// 返回值// 创建订单线程共享提交数据confirmVoThreadLocal.set(orderSubmitVO);// 1.生成订单实体对象(订单 + 订单项)OrderCreateTO order = this.createOrder();// 2.验价应付金额(允许0.01误差,前后端计算不一致)if (Math.abs(orderSubmitVO.getPayPrice().subtract(order.getPayPrice()).doubleValue()) >= 0.01) {// 验价不通过throw new VerifyPriceException();}// 验价成功// 3.保存订单saveOrder(order);// 4.库存锁定(wms_ware_sku)// 封装待锁定商品项TOWareSkuLockTO lockTO = new WareSkuLockTO();lockTO.setOrderSn(order.getOrder().getOrderSn());List<OrderItemVO> itemList = order.getOrderItems().stream().map((item) -> {OrderItemVO itemVO = new OrderItemVO();itemVO.setSkuId(item.getSkuId());itemVO.setCount(item.getSkuQuantity());itemVO.setTitle(item.getSkuName());return itemVO;}).collect(Collectors.toList());lockTO.setLocks(itemList);// 待锁定订单项R response = wmsFeignService.orderLockStock(lockTO);if (response.getCode() == 0) {// 锁定成功// TODO 5.远程扣减积分// 封装响应数据返回result.setOrder(order.getOrder());//System.out.println(10 / 0); // 模拟订单回滚,库存不会滚// 6.发送创建订单到延时队列rabbitTemplate.convertAndSend(MQConstant.order_event_exchange, MQConstant.order_create_routekey, order.getOrder());return result;} else {// 锁定失败throw new NoStockException("");}}
2. 在容器注入消息交换机、队列并进行绑定
@Configuration
public class MyRabbitMQConfig {/*** 延时队列*/@Beanpublic Queue orderDelayQueue() {/*** Queue(String name, 队列名字* boolean durable, 是否持久化* boolean exclusive, 是否排他* boolean autoDelete, 是否自动删除* Map<String, Object> arguments) 属性【TTL、死信路由、死信路由键】*/HashMap<String, Object> arguments = new HashMap<>();arguments.put("x-dead-letter-exchange", MQConstant.order_event_exchange);// 死信路由arguments.put("x-dead-letter-routing-key", MQConstant.order_release_routekey);// 死信路由键arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟return new Queue(MQConstant.order_delay_queue, true, false, false, arguments);}/*** 交换机(死信路由)*/@Beanpublic Exchange orderEventExchange() {return new TopicExchange(MQConstant.order_event_exchange, true, false);}/*** 死信队列*/@Beanpublic Queue orderReleaseQueue() {return new Queue(MQConstant.order_release_queue, true, false, false);}/*** 绑定:交换机与订单解锁延迟队列*/@Beanpublic Binding orderCreateBinding() {/*** String destination, 目的地(队列名或者交换机名字)* DestinationType destinationType, 目的地类型(Queue、Exhcange)* String exchange,* String routingKey,* Map<String, Object> arguments**/return new Binding(MQConstant.order_delay_queue,Binding.DestinationType.QUEUE,MQConstant.order_event_exchange,MQConstant.order_create_routekey,null);}/*** 绑定:交换机与订单解锁死信队列*/@Beanpublic Binding orderReleaseBinding() {return new Binding(MQConstant.order_release_queue,Binding.DestinationType.QUEUE,MQConstant.order_event_exchange,MQConstant.order_release_routekey,null);}/*** 绑定:交换机与库存解锁*/@Beanpublic Binding orderReleaseOtherBinding() {return new Binding(MQConstant.stock_release_queue,Binding.DestinationType.QUEUE,MQConstant.order_event_exchange,"order.release.other.#",null);}
}
3. 监听死信队列,进行关单,确认消息
@Slf4j
@RabbitListener(queues = MQConstant.order_release_queue)
@Component
public class OrderCloseListener {@AutowiredOrderService orderService;/*** 定时关单,监听死信队列,如果死信队列 消息过期时间 1分钟 后没有消费,就该关单* @param order* @param message* @param channel* @throws IOException*/@RabbitHandlerpublic void handleOrderRelease(OrderEntity order, Message message, Channel channel) throws IOException {log.debug("订单解锁,订单号:" + order.getOrderSn());try {orderService.closeOrder(order);// 手动删除消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 解锁失败 将消息重新放回队列,让别人消费channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);}}}
4. 关闭订单
只有是待付款状态才要关单
@Overridepublic void closeOrder(OrderEntity order) {OrderEntity _order = this.getById(order.getId());//只有是待付款状态才要关单if (OrderConstant.OrderStatusEnum.CREATE_NEW.getCode().equals(_order.getStatus())) {// 待付款状态允许关单OrderEntity temp = new OrderEntity();temp.setId(order.getId());temp.setStatus(OrderConstant.OrderStatusEnum.CANCLED.getCode());updateById(temp);try {// 发送消息给MQOrderTO orderTO = new OrderTO();BeanUtils.copyProperties(_order, orderTO);//TODO 持久化消息到mq_message表中,并设置消息状态为3-已抵达(保存日志记录)rabbitTemplate.convertAndSend(MQConstant.order_event_exchange, "order.release.other", orderTO);} catch (Exception e) {// TODO 消息为抵达Broker,修改mq_message消息状态为2-错误抵达}}}
五、源码下载
https://gitee.com/charlinchenlin/koo-erp