4.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队列，最终消费者拿到后会进行消费
利用死信交换机、死信队列。过期时间的方式，模拟出了延迟消息的效果
验证一下
- 在控制台创建simple.direct交换机
- 将此交换机与simple.queue绑定
注意！simple.queue并没有绑定到消费者，进入到simple.queue队列的消息都会变成死信
- 创建队列dlx.queue和dlx.direct并将其绑定
创建队列
创建交换机
进行绑定
- 给simple.queue队列设定死信交换机
注意，这个地方只能是在创建队列的时候进行绑定
- 在消费者模块代码中定义两个队列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 +"】");
}
- 发送消息
在控制台中下面的这个属性是带过期时间的属性
Java代码中的发送消息如下所示
@Test
void 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 keykey = "delay"
))
public void listenDelayMessage(String msg) {log.info("接收到delay.queue的延迟消息【" + msg + "】");
}
-
注入Bean的方式
这种方式只声明了交换机
@Bean
public DirectExchange delayExchange(){return ExchangeBuilder.directExchange("delay.direct").delayed() //设置delay的属性为true 主要是这个.durable(true) //持久化.build();
}
发送延迟消息的Java代码
@Testvoid testSendDelayMessage() {
// Message message = MessageBuilder
// .withBody("hello".getBytes(StandardCharsets.UTF_8))
// .setExpiration("10000") //过期时间10s
// .build();
// //发送到死信队列rabbitTemplate.convertAndSend("dela.direct", "hi", "hello", new MessagePostProcessor() {@Overridepublic 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;@Data
public 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
@RequiredArgsConstructor
public 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 = 1orderService.lambdaUpdate().set(Order::getStatus, 2).set(Order::getPayTime, LocalDateTime.now()).eq(Order::getId, orderId).eq(Order::getStatus, 1).update();}
}
在方法中发送延迟检查订单状态的消息
@Override@GlobalTransactionalpublic Long createOrder(OrderFormDTO orderFormDTO) {// 1.订单数据Order order = new Order();// 1.1.查询商品List<OrderDetailDTO> detailDTOS = orderFormDTO.getDetails();// 1.2.获取商品id和数量的MapMap<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.基于商品价格、购买数量计算商品总价：totalFeeint 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了
@RequiredArgsConstructor
public class DelayMessageProcessor implements MessagePostProcessor {private final int delay;@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDelay(delay);return message;}
}
或者是使用下面视频里面的代码
但是下面的代码每次都要使用一个内部类
