0


RabbitMQ 篇-深入了解延迟消息、MQ 可靠性(生产者可靠性、MQ 可靠性、消费者可靠性)

??博客主页:【_-CSDN博客】**
感谢大家点赞??收藏评论**

文章目录


1.0 RabbitMQ 的可靠性

消息从生产者到消费者的每一步都可能导致消息丢失:

1)发送消息时丢失:

- 生产者发送消息时连接 MQ 失败。

- 生产者发送消息到达 MQ 后未找到 Exchange 。

- 生产者发送消息到达 MQ 的 Exchange 后,未找到合适的 Queue 。

- 消息到达 MQ 后,处理消息的进程发生异常。

2)MQ 导致消息丢失:

- 消息到达 MQ,保存到队列后,尚未消费就突然宕机了。

3)消费者处理消息时:

- 消息接收后尚未处理突然宕机。

- 消息接收后处理过程中抛出异常。

综上,要解决消息丢失问题,保证 MQ 可靠性,就必须从三个方面入手:

- 确保生产者一定把消息发送到 MQ 。

- 确保 MQ 不会将消息弄丢。

- 确保消费者一定要处理消息。

2.0 发送者的可靠性

通过生产者重试机制、确认机制来确保发送者的可靠性。

2.1 生产者重试机制

首先第一种情况,就是生产者发送消息时,出现了网络故障,导致与MQ的连接中断。为了解决这个问题,SpringAMQP 提供的消息发送时的重试机制。即:当 RabbitTemplate 与 MQ 连接超时后,多次重试。

修改 application.yml 文件,添加下面的内容:

  1. spring:
  2. rabbitmq:
  3. connection-timeout: 1s # 设置MQ的连接超时时间
  4. template:
  5. retry:
  6. enabled: true # 开启超时重试机制
  7. initial-interval: 1000ms # 失败后的初始等待时间
  8. multiplier: 1 # 失败后下次的等待时长倍数,下次等待时长 = initial-interval * multiplier
  9. max-attempts: 3 # 最大重试次数

利用命令停掉 RabbitMQ 服务:

  1. docker stop mq

然后测试发送一条消息,会发现会每隔 1 秒重试 1 次,总共重试了 3 次。消息发送的超时重试机制配置成功了。

注意,当网络不稳定的时候,利用重试机制可以有效提高消息发送的成功率。不过 SpringAMQP 提供的重试机制时阻塞式的重试,也就是说多次重试等待的过程中,当前线程式被阻塞的。

如果对业务性能有要求,建议禁用重试机制,如果一定要使用,请合理配置等待时长和重试次数,当然也可以考虑使用异步线程来执行发送消息的代码。

2.2 生产者确认机制

一般情况下,只要生产者与 MQ 之间的网路连接顺畅,基本不会出现发送消息丢失的情况,因此大多数情况下我们无需考虑这种问题。

不过,在少数情况下,也会出现消息发送到 MQ 之后丢失的现象,比如:

- MQ 内部处理消息的进程发生了异常。

- 生产者发送消息到达 MQ 后未找到 Exchange 。

- 生产者发送消息到达 MQ 的 Exchange 后,未找到合适的 Queue,因此无法路由。

当消息投递到 MQ,但是路由失败时,通过 Return 返回异常信息,同时返回 ack 确认信息,代表投递成功。

1)临时消息投递到了 MQ,并且入队成功,返回 ACK,告知投递成功。

2)持久消息投递到了 MQ,并且入队完成持久化,返回 ACK,告知投递成功。

3)其他情况都会返回 NACK,告知投递失败。

其中 ACK 和 NACK 属于Publisher Confirm机制,ACK是投递成功;NACK 是投递失败。而 return 则属于Publisher Return机制。

2.2.1 开启生产者确认机制

默认两种机制都是关闭状态,需要通过配置文件来开启。

在 application.yml 中添加配置:

  1. spring:
  2. rabbitmq:
  3. publisher-confirm-type: correlated # 开启publisher confirm机制,并设置confirm类型
  4. publisher-returns: true # 开启publisher return机制

这里 publisher-confirm-type 有三种模式可选:

1)none:关闭 confirm 机制

2)simple:同步阻塞等待 MQ 的回执

3)correlated:MQ 异步回调返回回执

一般我们推荐使用 correlated,回调机制。

2.2.2 定义 ReturnCallback 机制

每个 RabbitTemplate 只能配置一个 ReturnCallback,因此可以在配置类中统一设置。

代码如下:

  1. package com.itheima.publisher.config;
  2. import lombok.AllArgsConstructor;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.core.ReturnedMessage;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.context.annotation.Configuration;
  7. import javax.annotation.PostConstruct;
  8. @Slf4j
  9. @AllArgsConstructor
  10. @Configuration
  11. public class MqConfig {
  12. private final RabbitTemplate rabbitTemplate;
  13. @PostConstruct
  14. public void init(){
  15. rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
  16. @Override
  17. public void returnedMessage(ReturnedMessage returned) {
  18. log.error("触发return callback,");
  19. log.debug("exchange: {}", returned.getExchange());
  20. log.debug("routingKey: {}", returned.getRoutingKey());
  21. log.debug("message: {}", returned.getMessage());
  22. log.debug("replyCode: {}", returned.getReplyCode());
  23. log.debug("replyText: {}", returned.getReplyText());
  24. }
  25. });
  26. }
  27. }

测试:

当发送消息的时候,设置一个不存在的路由:

在交换机中,不存在 “xbss” 路由关键字,则会执行returnedMessage 方法。

  1. // 首先引入依赖
  2. @Autowired
  3. private RabbitTemplate rabbitTemplate;
  4. @Test
  5. void contextLoads() {
  6. //发送Object类型的消息
  7. textObject();
  8. }
  9. @Test
  10. public void textObject(){
  11. User user = new User(1,"xbs","123456",null,null);
  12. rabbitTemplate.convertAndSend("textExchange", "xbss", user, new MessagePostProcessor() {
  13. @Override
  14. public Message postProcessMessage(Message message) throws AmqpException {
  15. message.getMessageProperties().setMessageId("123456");
  16. return message;
  17. }
  18. });
  19. }

执行结果:

2.2.3 定义 ConfirmCallback 机制

由于每个消息发送时的处理逻辑不一定相同,因此 ConfirmCallback 需要在每次发送消息时定义。具体来说,是在调用 RabbitTemplate 中的 convertAndSend 方法时,多传递一个参数:

代码如下:

  1. @Test
  2. void testPublisherConfirm() {
  3. // 1.创建CorrelationData
  4. CorrelationData cd = new CorrelationData();
  5. // 2.给Future添加ConfirmCallback
  6. cd.getFuture().thenAccept(result -> {
  7. if (result.isAck()) { // result.isAck(),boolean类型,true代表ack回执,false 代表 nack回执
  8. log.info("发送消息成功,收到 ack!");
  9. } else { // result.getReason(),String类型,返回nack时的异常描述
  10. log.info("发送消息失败,收到 nack, reason : {}", result.getReason());
  11. }
  12. }).exceptionally(ex -> {
  13. // 2.1.Future发生异常时的处理逻辑,基本不会触发
  14. log.error("send message fail", ex);
  15. return null;
  16. });
  17. // 3.发送消息
  18. rabbitTemplate.convertAndSend("xbs.direct", "red", "hello", cd);
  19. }

执行结果:

执行成功之后,日志输出:

接收消息:

注意

开启生产者确认比较消耗 MQ 性能,一般不建议开启。而且大家思考一下触发确认的几种情况:

  • 路由失败:一般是因为 RoutingKey 错误导致,往往是编程导致
  • 交换机名称错误:同样是编程错误导致
  • MQ 内部故障:这种需要处理,但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启,而且仅仅需要开启 ConfirmCallback 处理 nack 就可以了。

3.0 MQ 可靠性

消息到达MQ以后,如果 MQ 不能及时保存,也会导致消息丢失,所以 MQ 的可靠性也非常重要。

为了提升性能,默认情况下 MQ 的数据都是在内存存储的临时数据,重启后就会消失。为了保证数据的可靠性,必须配置数据持久化,包括:

1)交换机持久化

2)队列持久化

3)消息持久化

3.1 交换机持久化

在控制台的 Exchange 页面,添加交换机时可以配置交换机的 Durability 参数:

设置为 Durable 就是持久化模式,Transient 就是临时模式。

3.2 队列持久化

在控制台的 Queues 页面,添加队列时,同样可以配置队列的 Durability 参数:

设置为 Durable 就是持久化模式,Transient 就是临时模式。

3.3 消息持久化

在控制台发送消息的时候,可以添加很多参数,而消息的持久化是要配置一个Delivery mode:设置为 Persistent 消息持久化。

说明

在开启持久化机制以后,如果同时还开启了生产者确认,那么 MQ 会在消息持久化以后才发送 ACK 回执,进一步确保消息的可靠性。

不过出于性能考虑,为了减少 IO 次数,发送到 MQ 的消息并不是逐条持久化到数据库的,而是每隔一段时间批量持久化。一般间隔在 100 毫秒左右,这就会导致 ACK 有一定的延迟,因此建议生产者确认全部采用异步方式。

3.4 LazyQueue 懒惰队列

1)在默认情况下,RabbitMQ 会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下,这会导致消息积压,比如:

  • 消费者宕机或出现网络故障
  • 消息发送量激增,超过了消费者处理速度
  • 消费者处理业务发生阻塞

2)一旦出现消息堆积问题,RabbitMQ 的内存占用就会越来越高,直到触发内存预警上限。此时 RabbitMQ 会将内存消息刷到磁盘上,这个行为成为 PageOut 。PageOut 会耗费一段时间,并且会阻塞队列进程。因此在这个过程中 RabbitMQ 不会再处理新的消息,生产者的所有请求都会被阻塞。

为了解决这个问题,从 RabbitMQ 的 3.6.0 版本开始,就增加了 Lazy Queues 的模式,也就是惰性队列。惰性队列的特征如下:

  • 接收到消息后直接存入磁盘而非内存
  • 消费者要消费消息时才会从磁盘中读取并加载到内存(也就是懒加载)
  • 支持数百万条的消息存储

而在 3.12 版本之后,LazyQueue 已经成为所有队列的默认格式。因此官方推荐升级 MQ 为 3.12 版本或者所有队列都设置为 LazyQueue 模式。

3.4.1 声明懒惰队列的方式

1)在添加队列的时候,添加 x-queue-mode参数即可设置队列为 Lazy 模式:

2)在利用 SpringAMQP 声明队列的时候,添加 x-queue-mode 参数也可设置队列为 Lazy 模式:

  1. @Bean
  2. public Queue queue3(){
  3. return QueueBuilder
  4. .durable("queue7") //队列名称
  5. .lazy() //开启懒惰模式
  6. .build();
  7. }

3)基于注解来声明队列并设置为 Lazy 模式:

  1. @RabbitListener(queuesToDeclare = @Queue(
  2. name = "lazy.queue",
  3. durable = "true",
  4. arguments = @Argument(name = "x-queue-mode", value = "lazy")
  5. ))
  6. public void listenLazyQueue(String msg){
  7. log.info("接收到 lazy.queue的消息:{}", msg);
  8. }

4.0 消费者的可靠性

当 RabbitMQ 向消费者投递消息以后,需要知道消费者的处理状态如何。因为消息投递给消费者并不代表就一定被正确消费了,可能出现的故障有很多。

一旦发生上述情况,消息也会丢失。因此,RabbitMQ 必须知道消费者的处理状态,一旦消息处理失败才能重新投递消息。

4.1 消费者确认机制

1)为了确认消费者是否成功处理消息,RabbitMQ 提供了消费者确认机制(Consumer Acknowledgement)。即:当消费者处理消息结束后,应该向 RabbitMQ 发送一个回执,告知 RabbitMQ 自己消息处理状态。回执有三种可选值:

  • ack:成功处理消息,RabbitMQ 从队列中删除该消息。
  • nack:消息处理失败,RabbitMQ 需要再次投递消息。
  • reject:消息处理失败并拒绝该消息,RabbitMQ 从队列中删除该消息。

2)一般 reject 方式用的较少,除非是消息格式有问题,那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过 try catch 机制捕获,消息处理成功时返回 ack,处理失败时返回 nack 。

由于消息回执的处理代码比较统一,因此 SpringAMQP 帮我们实现了消息确认。并允许我们通过配置文件设置 ACK 处理方式,有三种模式:

  • none:不处理。即消息投递给消费者后立刻 ack,消息会立刻从 MQ 删除。非常不安全,不建议使用
  • manual:手动模式。需要自己在业务代码中调用 api,发送 ack 或 reject,存在业务入侵,但更灵活
  • auto:自动模式。SpringAMQP 利用 AOP 对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回 ack. 当业务出现异常时,根据异常判断返回不同结果:- 如果是业务异常,会自动返回 nack;- 如果是消息处理或校验异常,自动返回 reject;

因此,推荐使用 aotu 自动模式。

application 配置如下:

  1. spring:
  2. rabbitmq:
  3. listener:
  4. simple:
  5. acknowledge-mode: auto # 自动ack

在开启消费者确认机制时,手动抛出异常模拟接收信息失败时:

1)发送消息:

  1. // 首先引入依赖
  2. @Autowired
  3. private RabbitTemplate rabbitTemplate;
  4. @Test
  5. void contextLoads() {
  6. //发送Object类型的消息
  7. testSendMessage();
  8. }
  9. @Test
  10. public void testSendMessage(){
  11. //发送消息
  12. rabbitTemplate.convertAndSend("yt","xbs","hello rabbitmq");
  13. }

2)接收消息:

  1. @RabbitListener(bindings = @QueueBinding(
  2. value = @Queue(name = "wyt",durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy")),
  3. exchange = @Exchange(name = "yt",type = ExchangeTypes.DIRECT),
  4. key = "xbs"
  5. ))
  6. public void receiveMessage(String massage) throws InterruptedException {
  7. log.info("sendMessage发送的消息为: " + massage);
  8. //模拟接送消息失败
  9. throw new RuntimeException("模拟接收消息失败");
  10. }

执行结果:

由于抛出了异常,则返回给 MQ 为 nack,就会重复发送消息给客户端:

在队列中,由于消息没有被正确处理,则消息会一直在队列中且不断发送消息给客户端:

4.2 失败重试机制

当消费者出现异常后,消息会不断 requeue(重入队)到队列,再重新发送给消费者。如果消费者再次执行依然出错,消息会再次 requeue 到队列,再次投递,直到消息处理成功为止。

极端情况就是消费者一直无法执行成功,那么消息 requeue 就会无限循环,导致mq的消息处理飙升,带来不必要的压力:

应对上述情况 Spring 又提供了消费者失败重试机制:在消费者出现异常时利用本地重试,而不是无限制的 requeue 到 mq 队列。

修改 application.yml 文件,添加内容:

  1. spring:
  2. rabbitmq:
  3. listener:
  4. simple:
  5. retry:
  6. enabled: true # 开启消费者失败重试
  7. initial-interval: 1000ms # 初识的失败等待时长为1秒
  8. multiplier: 1 # 失败的等待时长倍数,下次等待时长 = multiplier * last-interval
  9. max-attempts: 3 # 最大重试次数
  10. stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为false

再来尝试手动抛出异常模拟接收消息失败:

结果如下:

MQ 向接收者发送了三次信息,接着进程结束。

此时队列中的信息不存在了,丢失了:

结论:

  • 开启本地重试时,消息处理过程中抛出异常,不会 requeue 到队列,而是在消费者本地重试
  • 重试达到最大次数后,Spring 会返回 reject,消息会被丢弃。

4.3 失败处理策略

在之前的测试中,本地测试达到最大重试次数后,消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下,显然不太合适了。

因此 Spring 允许我们自定义重试次数耗尽后的消息处理策略,这个策略是由 MessageRecovery 接口来定义的,它有3个不同实现:

  • RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息。默认就是这种方式
  • ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队
  • RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机

**比较优雅的一种处理方案是

  1. RepublishMessageRecoverer

,失败后将消息投递到一个指定的,专门存放异常消息的队列,后续由人工集中处理。**

代码实现:

1)定义处理失败消息的交换机和队列:

  1. @Bean
  2. public DirectExchange errorMessageExchange(){
  3. return new DirectExchange("error.direct");
  4. }
  5. @Bean
  6. public Queue errorQueue(){
  7. return new Queue("error.queue", true);
  8. }
  9. @Bean
  10. public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
  11. return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
  12. }

2)定义一个 RepublishMessageRecoverer,关联队列和交换机

  1. @Bean
  2. public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
  3. return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
  4. }

完整代码:

  1. package com.itheima.consumer.config;
  2. import org.springframework.amqp.core.Binding;
  3. import org.springframework.amqp.core.BindingBuilder;
  4. import org.springframework.amqp.core.DirectExchange;
  5. import org.springframework.amqp.core.Queue;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.amqp.rabbit.retry.MessageRecoverer;
  8. import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
  9. import org.springframework.context.annotation.Bean;
  10. @Configuration
  11. @ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
  12. public class ErrorMessageConfig {
  13. @Bean
  14. public DirectExchange errorMessageExchange(){
  15. return new DirectExchange("error.direct");
  16. }
  17. @Bean
  18. public Queue errorQueue(){
  19. return new Queue("error.queue", true);
  20. }
  21. @Bean
  22. public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
  23. return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
  24. }
  25. @Bean
  26. public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
  27. return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
  28. }
  29. }

当重试失败之后,不会直接将消息丢失而是交给专门接收失败信息的交换机,再由专门的消费者进行消费,比如说监听到队列中有接收失败的信息,将其写入日志中等等处理的方法。

测试:

1)发送消息:

  1. // 首先引入依赖
  2. @Autowired
  3. private RabbitTemplate rabbitTemplate;
  4. @Test
  5. void contextLoads() {
  6. //发送Object类型的消息
  7. testSendMessage();
  8. }
  9. @Test
  10. public void testSendMessage(){
  11. //发送消息
  12. rabbitTemplate.convertAndSend("yt","xbs","hello rabbitmq");
  13. }

2)接收消息:

接收消息之后,手动抛出异常模拟接收消息失败。

  1. @RabbitListener(bindings = @QueueBinding(
  2. value = @Queue(name = "wyt",durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy")),
  3. exchange = @Exchange(name = "yt",type = ExchangeTypes.DIRECT),
  4. key = "xbs"
  5. ))
  6. public void receiveMessage(String massage) throws InterruptedException {
  7. log.info("sendMessage发送的消息为: " + massage);
  8. //模拟接送消息失败
  9. throw new RuntimeException("模拟接收消息失败");
  10. }

执行结果:

在本地重试三次发送失败之后,则将该消息交给 “errorMessageExchange” 交换机,再路由给 “errorMessageQueue” 队列,后续可以监听该队列进行接收消息再做处理。

5.0 使用 DelayExchange 插件

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

5.1 安装 DelayExchange 插件

基于 Docker 安装,所以需要先查看 RabbitMQ 的插件目录对应的数据卷。

  1. docker volume inspect mq-plugins

结果如下:

  1. [
  2. {
  3. "CreatedAt": "2024-06-19T09:22:59+08:00",
  4. "Driver": "local",
  5. "Labels": null,
  6. "Mountpoint": "/var/lib/docker/volumes/mq-plugins/_data",
  7. "Name": "mq-plugins",
  8. "Options": null,
  9. "Scope": "local"
  10. }
  11. ]

插件目录被挂载到了 docker volume inspect mq-plugins 这个目录,我们上传插件到该目录下。

DelayExchange 插件下载地址:https://github.com/rabbitmq/rabbitmq-delayed-message-exchange

接着文件放到docker volume inspect mq-plugins 这个目录中。

接下来执行命令,安装插件:

  1. docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

安装结果:

5.2 声明延迟交换机

1)基于注解方式:

  1. @RabbitListener(bindings = @QueueBinding(
  2. value = @Queue(name = "delay.queue", durable = "true"),
  3. exchange = @Exchange(name = "delay.direct", delayed = "true"),
  4. key = "delay"
  5. ))
  6. public void listenDelayMessage(String msg){
  7. log.info("接收到delay.queue的延迟消息:{}", msg);
  8. }

2)基于 @Bean 的方式:

  1. package com.itheima.consumer.config;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6. @Slf4j
  7. @Configuration
  8. public class DelayExchangeConfig {
  9. @Bean
  10. public DirectExchange delayExchange(){
  11. return ExchangeBuilder
  12. .directExchange("delay.direct") // 指定交换机类型和名称
  13. .delayed() // 设置delay的属性为true
  14. .durable(true) // 持久化
  15. .build();
  16. }
  17. @Bean
  18. public Queue delayedQueue(){
  19. return new Queue("delay.queue");
  20. }
  21. @Bean
  22. public Binding delayQueueBinding(){
  23. return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
  24. }
  25. }

5.3 发送延迟消息

发送消息时,必须通过 x-delay 属性设定延迟时间:

  1. @Test
  2. void testPublisherDelayMessage() {
  3. // 1.创建消息
  4. String message = "hello, delayed message";
  5. // 2.发送消息,利用消息后置处理器添加消息头
  6. rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {
  7. @Override
  8. public Message postProcessMessage(Message message) throws AmqpException {
  9. // 添加延迟消息属性
  10. message.getMessageProperties().setDelay(5000);
  11. return message;
  12. }
  13. });
  14. }

测试:

1)接收消息:

  1. @Autowired
  2. MessageConverter messageConverter;
  3. @RabbitListener(bindings = @QueueBinding(
  4. value = @Queue(name = "delay.queue",durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy")),
  5. exchange = @Exchange(name = "delay.exchange",type = ExchangeTypes.DIRECT,delayed = "true"),
  6. key = "xbs"
  7. ))
  8. public void receiveMessage(Message massage) throws InterruptedException {
  9. log.info("sendMessage发送的消息ID为: " + massage.getMessageProperties().getMessageId());
  10. log.info("sendMessage发送的消息延迟时间为: " + massage.getMessageProperties().getDelayLong());
  11. String str = (String) messageConverter.fromMessage(massage);
  12. log.info("sendMessage发送的消息内容为: " + str);
  13. }

2)发送消息:

  1. // 首先引入依赖
  2. @Autowired
  3. private RabbitTemplate rabbitTemplate;
  4. @Test
  5. void contextLoads() {
  6. //发送Object类型的消息
  7. testSendDelayMessage();
  8. }
  9. @Test
  10. void testSendDelayMessage(){
  11. //发送消息
  12. rabbitTemplate.convertAndSend("delay.exchange", "xbs", "hello rabbitmq delay message", new MessagePostProcessor() {
  13. @Override
  14. public Message postProcessMessage(Message message) throws AmqpException {
  15. //先设置唯一ID
  16. message.getMessageProperties().setMessageId("123456");
  17. //设置延迟发送时间
  18. message.getMessageProperties().setDelayLong(5000L);
  19. return message;
  20. }
  21. });
  22. }

执行结果:

注意:

延迟消息插件内部会维护一个本地数据库表,同时使用 Elang Timers 功能实现计时。如果消息的延迟时间设置较长,可能会导致堆积的延迟消息非常多,会带来较大的 CPU 开销,同时延迟消息的时间会存在误差。

因此,不建议设置延迟时间过长的延迟消息。


本文转载自: https://blog.csdn.net/web13093320398/article/details/143997703
版权归原作者 web13093320398 所有, 如有侵权,请联系我们删除。

“RabbitMQ 篇-深入了解延迟消息、MQ 可靠性(生产者可靠性、MQ 可靠性、消费者可靠性)”的评论:

还没有评论