第十六章 RabbitMQ延迟消息之延迟插件优化

目录

一、引言

二、优化方案 

三、核心代码实现

3.1. 生产者代码

3.2. 消息处理器 

3.3. 自定义多延迟消息封装类 

3.4. 订单实体类 

3.5. 消费者代码 

四、运行效果

---

一、引言

上一章节我们提到，直接使用延迟插件，创建一个延迟指定时间的消息（如10分钟），并不是最好的解决方案，因为假如我们的订单是在5分钟支付的，那么剩余的5分钟时间，RabbitMQ中延迟消息时钟还是一直占用着资源。如果有大量的延迟消息，那么对于服务来说压力是很大的，同时会耗费庞大昂贵的资源。因此，本章节我们就来近一步对延迟插件的消息进行优化。

我们通过下面的流程图来做近一步分析：

1. 用户下单完成后，发送15分钟延迟消息，在15分钟后接收消息，检查支付状态：

2. 已支付：更新订单状态为已支付

3. 未支付：更新订单状态为关闭订单，恢复商品库存

常规延迟插件消息使用的弊端总结：

1. 设置30分钟后检测订单支付状态实现起来非常简单，但是存在两个问题：

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

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

二、优化方案 

如下图所示，我们可以将10分钟甚至30分钟拆分成多份零散的较短的时间。

消息初次发送的延迟时间设定为10s，10s过后如果订单还是未支付状态，我们判断延迟时间数组里还有没有剩余延迟时间，如果有则继续发送延迟消息，时间设定为数组中的第二个时间10s，直到订单支付成功终止循环，或是最后一份时间消耗完依然未支付，我们取消订单。

三、核心代码实现

3.1. 生产者代码

package com.example.publisher;import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;/*** 生产者*/
@Slf4j
@SpringBootTest
class PublisherApplicationTests {@Resourceprivate RabbitTemplate rabbitTemplate;@Testvoid test() {Order order = Order.builder().orderId(1L).content("生活不易，所以保持足够的努力，对自己要有信心，积极地去面对工作生活的挑战！").build();MultiDelayMessage<Order> msg = MultiDelayMessage.of(order, 1000L, 5000L, 2000L, 10000L);rabbitTemplate.convertAndSend("delay.direct", "delay", msg, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDelayLong(msg.removeNextDelay());return message;}});}
}

3.2. 消息处理器 

package com.example.publisher;import lombok.AllArgsConstructor;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;/*** 消息请求处理器*/
@AllArgsConstructor
public class DelayMessageProcessor implements MessagePostProcessor {private final Long delay;@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDelayLong(delay);return message;}
}

3.3. 自定义多延迟消息封装类 

package com.example.publisher;import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.util.CollectionUtils;import java.io.Serializable;
import java.util.List;/*** 自定义的多延时消息封装类* @param <T>*/
@Data
@NoArgsConstructor
public class MultiDelayMessage<T> implements Serializable {/*** 消息体*/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, (List<Long>) CollectionUtils.arrayToList(delayMillis));}/*** 获取并移除下一个延迟时间* @return 队列中的第一个延迟时间*/public Long removeNextDelay() {return delayMillis.remove(0);}/*** 是否还有下一个延迟时间* @return*/public boolean hasNextDelay() {return !delayMillis.isEmpty();}
}

3.4. 订单实体类 

package com.example.publisher;import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;/*** 订单类 * 此处为了演示，将真实业务中的订单类做了简化* 只包含一个订单ID和自定义消息内容*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class Order {private Long orderId;private String content;
}

3.5. 消费者代码 

package com.example.consumer;import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;/*** 消费者* 因为作为演示，所以商城支付、订单、及扣减库存的业务代码已注释* 注释中保留了整个商城下单支付扣减库存的流程步骤*/
@Slf4j
@Component
public class SimpleListener {@Resourceprivate RabbitTemplate rabbitTemplate;@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "delay.queue", durable = "true"),exchange = @Exchange(name = "delay.direct", delayed = "true"),key = "delay"))public void listener(MultiDelayMessage<Order> msg) throws Exception {System.out.println(((Order)msg.getData()).getContent());// 1. 查询订单状态// Order order = orderService.getById(msg.getData())// 2. 判断是否已支付
//        if (Order == null || order.status == 2) {
//            订单不存在或者已处理则直接返回
//            return;
//        }// 主动去支付服务查询真正的支付状态
//        PayOrder payOrder = payService.getById(order.getId());// 2.1. 已支付，则标记订单为已支付
//        if (payOrder.isPay()) {
//            orderService.markOrderPaySuccess(order.getId());
//            return;
//        }// 2.2. 未支付，获取下次订单延迟时间// 3. 判断是否存在延迟时间if (msg.hasNextDelay()) {// 3.1 存在，重发延迟消息Long nextDelay = msg.removeNextDelay();rabbitTemplate.convertAndSend("delay.direct", "delay", msg, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDelayLong(nextDelay);return message;}});return;}// 3.2 不存在，取消订单
//        orderService.lambdaUpdate()
//                .set(Order::getStatus, 5);
//                .set(Order::getCloseTime, LocalDateTime.now());
//                .eq(Order::getId, order.getId())
//                .update();// 4. 恢复库存}
}

四、运行效果

最终我们会看到每间隔一段时间消费者就会消费一条消息，这个间隔时间就是我们设定的分段时间数组，这么做就能极大地减少资源消耗和服务的压力：