4.RabbitMQ - 延迟消息_rabbitmq延迟消息

RabbitMQ延迟消息

文章目录

• RabbitMQ延迟消息
• 一、延迟消息介绍
• 二、实现
• • 2.1 死信交换机
  • 2.2 延迟消息插件
  • 2.3 取消超时订单

一、延迟消息介绍

延迟消息：生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间后才收到消息

用户下单抢购，抢到了但是没有付款，此时其实库存的数量已经扣减了

如果用户迟迟没有付款，超过一定的时间，就会将此订单取消掉，库存的数量也会重新加回来

我们可以定义一个定时任务扫描数据中订单的状态，超过一定时间没有付款的，我们就将订单取消

延迟任务：设置在一定时间之后才执行的任务

当用户下单成功后，立刻向MQ中发送一条延迟消息，设定延迟时间30分钟，30分钟到了之后就可以收到此消息，检查订单状态，如果发现未支付，则订单直接取消。

这样解决了实效性的问题，同时对数据库的压力也很小

二、实现

2.1 死信交换机

当队列满足下列的条件之一时就会称为死信（dead letter）

• 消费者使用basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false

  消费者不要这个消息了

• 消息是一个过期的消息（达到了队列或消息本身设置的过期时间），超时无人消费

• 要投递的队列消息堆积满了，最早的消息可能成为死信

如果队列通过dead-letter-exchange属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中。这个交换机称为死信交换机（Dead Letter Exchange，简称DLX）

死信交换机只是一种称呼，和普通的交换机其实是一样的

我们不给simple.queue队列绑定消费者，给dlx.queue绑定一个消费者

因为simple.queue队列没有消费者，所以不会有人来消费，当有人通过simple.direct交换机向simple.queue队列发送一条过期时间为30秒的消息，此消息就会在simple.queue队列卡主

过了30s后，消息就会自动投递到dlx.direct死信交换机，然后进入dlx.queue队列，最终消费者拿到后会进行消费

利用死信交换机、死信队列。过期时间的方式，模拟出了延迟消息的效果

验证一下

1. 在控制台创建simple.direct交换机

2. 将此交换机与simple.queue绑定

注意！simple.queue并没有绑定到消费者，进入到simple.queue队列的消息都会变成死信

3. 创建队列dlx.queue和dlx.direct并将其绑定

创建队列

创建交换机

进行绑定

4. 给simple.queue队列设定死信交换机

注意，这个地方只能是在创建队列的时候进行绑定

5. 在消费者模块代码中定义两个队列simple.queue、dlx.queue

//检查一下，一定不要有simple.queue的消费者//@RabbitListener(queues = \"simple.queue\")//public void listenSimpleQueue(String msg){// System.out.println(\"消费者收到了simple.queue的消息：【\" + msg +\"】\");// throw new RuntimeException(\"抛出异常了\");//}@RabbitListener(queues = \"dlx.queue\")public void listenDlxQueue(String msg){ log.info(\"消费者收到了dlx.queue的消息：【\" + msg +\"】\");}

6. 发送消息

在控制台中下面的这个属性是带过期时间的属性

Java代码中的发送消息如下所示

@Testvoid testSendTTLMessage() { Message message = MessageBuilder .withBody(\"hello\".getBytes(StandardCharsets.UTF_8)) .setExpiration(\"10000\") //过期时间10s .build(); //发送到死信队列 rabbitTemplate.convertAndSend(\"simple.direct\", \"hi\", message);//直接向队列发送消息 log.info(\"消息发送成功！\");}

simple.queue队列中10s内始终存在下面的一条消息

dlx.queue队列的消费者在10s后会接收到消息

2.2 延迟消息插件

这种定时功能，都是有一定的性能损耗的（Redis除外）

MQ或者Spring的定时功能是在程序内部维护一个时钟，比如每隔一秒就往前跳一次，这种时钟的运行过程中CPU就需要不停地计算，定时任务越多，对于CPU的占用越大，定时任务属于一种CPU密集型的任务

采用延迟消息带来的弊端就是给服务器CPU造成的额外压力比较大

使用交换机实现延迟消息非常的繁琐，需要定义很多的交换机和队列，而且死信交换机的目的是为了让我们人工处理死信消息，并不是为了延迟消息而生的

延迟消息的插件能自动实现延迟效果

RabbitMQ官方也推出了一个插件，原生支持延迟消息功能。

该插件的原理是设计了一种支持延迟消息功能的交换机，当消息投递到交换机后可以暂存一定时间，到期后再投递到队列。

暂存的时间取决于发消息时配置的时间（也就是延迟时间）

在Java代码中配置延迟交换机的两种方式

在声明交换机的时候，需要多添加一个参数delayed=“true”

• 注解的方式

  在消费者模块声明交换机、队列

@RabbitListener(bindings = @QueueBinding( //队列 value = @Queue(name = \"delay.queue\", durable = \"true\"), //交换机 exchange = @Exchange(name = \"hmall.direct\", delayed = \"true\"), //Routing key key = \"delay\"))public void listenDelayMessage(String msg) { log.info(\"接收到delay.queue的延迟消息【\" + msg + \"】\");}

• 注入Bean的方式

  这种方式只声明了交换机

@Beanpublic DirectExchange delayExchange(){ return ExchangeBuilder .directExchange(\"delay.direct\") .delayed() //设置delay的属性为true 主要是这个 .durable(true) //持久化 .build();}

发送延迟消息的Java代码

 @Test void testSendDelayMessage() {// Message message = MessageBuilder// .withBody(\"hello\".getBytes(StandardCharsets.UTF_8))// .setExpiration(\"10000\") //过期时间10s// .build();// //发送到死信队列 rabbitTemplate.convertAndSend(\"dela.direct\", \"hi\", \"hello\", new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration(\"10000\");//延迟十秒 return message; } });//直接向队列发送消息 log.info(\"消息发送成功！\"); }

2.3 取消超时订单

设置三十分钟后检测订单支付状态，存在两个问题

• 如果并发比较高，30分钟可能堆积消息过多，对MQ的压力很大

• 大多数订单在下单后1分钟内就会支付，但是却需要再MQ内等待30分钟，浪费资源

  30分钟太长，可以缩短为10s，10s后立刻来检查有没有支付

  假如10s后没有支付，可以再发一个10s的延迟消息，直到成功后不再发送延迟消息

  这样的话MQ的压力会减少很多

处理如下所示：

查询支付状态的时候，需要先查询本地，之后再查询支付服务，查完之后判断支付状态

定义延时消息时间数组

package com.hmall.common.domain;import com.hmall.common.utils.CollUtils;import lombok.Data;import java.util.List;@Datapublic class MultiDelayMessage<T> { /** * 消息体 */ private T data; /** * 记录延迟时间的集合 */ private List<Long> delayMillis; public MultiDelayMessage(T data, List<Long> delayMillis) { this.data = data; this.delayMillis = delayMillis; } public static <T> MultiDelayMessage<T> of(T data, Long ... delayMillis){ return new MultiDelayMessage<>(data, CollUtils.newArrayList(delayMillis)); } /** * 获取并移除下一个延迟时间 * @return 队列中的第一个延迟时间 */ public Long removeNextDelay(){ return delayMillis.remove(0); } /** * 是否还有下一个延迟时间 */ public boolean hasNextDelay(){ return !delayMillis.isEmpty(); }}

定义好对应的交换机和队列

@Component@RequiredArgsConstructorpublic class PayStatusListener { private final IOrderService orderService; @RabbitListener(bindings = @QueueBinding( value = @Queue(name = \"mark.order.pay.queue\", durable = \"true\"), exchange = @Exchange(name = \"pay.topic\", type = ExchangeTypes.TOPIC), key = \"pay.success\" )) public void listenOrderPay(Long orderId) { /* // 1.查询订单 Order order = orderService.getById(orderId); // 2.判断订单状态是否为未支付 if(order == null || order.getStatus() != 1){ // 订单不存在，或者状态异常 return; } // 3.如果未支付，标记订单状态为已支付 orderService.markOrderPaySuccess(orderId);*/ // update order set status = 2 where id = ? AND status = 1 orderService.lambdaUpdate() .set(Order::getStatus, 2) .set(Order::getPayTime, LocalDateTime.now()) .eq(Order::getId, orderId) .eq(Order::getStatus, 1) .update(); }}

在方法中发送延迟检查订单状态的消息

 @Override @GlobalTransactional public Long createOrder(OrderFormDTO orderFormDTO) { // 1.订单数据 Order order = new Order(); // 1.1.查询商品 List<OrderDetailDTO> detailDTOS = orderFormDTO.getDetails(); // 1.2.获取商品id和数量的Map Map<Long, Integer> itemNumMap = detailDTOS.stream() .collect(Collectors.toMap(OrderDetailDTO::getItemId, OrderDetailDTO::getNum)); Set<Long> itemIds = itemNumMap.keySet(); // 1.3.查询商品 List<ItemDTO> items = itemClient.queryItemByIds(itemIds); if (items == null || items.size() < itemIds.size()) { throw new BadRequestException(\"商品不存在\"); } // 1.4.基于商品价格、购买数量计算商品总价：totalFee int total = 0; for (ItemDTO item : items) { total += item.getPrice() * itemNumMap.get(item.getId()); } order.setTotalFee(total); // 1.5.其它属性 order.setPaymentType(orderFormDTO.getPaymentType()); order.setUserId(UserContext.getUser()); order.setStatus(1); // 1.6.将Order写入数据库order表中 save(order); // 2.保存订单详情 List<OrderDetail> details = buildDetails(order.getId(), items, itemNumMap); detailService.saveBatch(details); // 3.扣减库存 try { itemClient.deductStock(detailDTOS); } catch (Exception e) { throw new RuntimeException(\"库存不足！\"); } // 4.清理购物车商品 // cartClient.deleteCartItemByIds(itemIds); try { rabbitTemplate.convertAndSend(  MqConstants.TRADE_EXCHANGE_NAME, MqConstants.ORDER_CREATE_KEY,  itemIds/*,  new RelyUserInfoMessageProcessor()*/ ); } catch (AmqpException e) { log.error(\"清理购物车的消息发送异常\", e); } // 5.延迟检测订单状态消息 try { MultiDelayMessage<Long> msg = MultiDelayMessage.of(order.getId(), 10000L, 10000L, 10000L, 15000L, 15000L, 30000L, 30000L); rabbitTemplate.convertAndSend(  MqConstants.DELAY_EXCHANGE, MqConstants.DELAY_ORDER_ROUTING_KEY, msg,  new DelayMessageProcessor(msg.removeNextDelay().intValue()) ); } catch (AmqpException e) { log.error(\"延迟消息发送异常！\", e); } return order.getId(); }

将MessagePostProcessMessage对象提取出来了，不用每次都new了

@RequiredArgsConstructorpublic class DelayMessageProcessor implements MessagePostProcessor { private final int delay; @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setDelay(delay); return message; }}

或者是使用下面视频里面的代码

但是下面的代码每次都要使用一个内部类