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

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

文章目录

???1.0 RabbitMQ 的可靠性

? ? ? ? 2.0 发送者的可靠性

? ? ? ? 2.1 生产者重试机制

? ? ? ? 2.2 生产者确认机制

? ? ? ? 2.2.1 开启生产者确认机制

? ? ? ? 2.2.2 定义 ReturnCallback 机制

? ? ? ? 2.2.3 定义 ConfirmCallback 机制

???3.0 MQ 可靠性

? ? ? ? 3.1 交换机持久化

? ? ? ? 3.2 队列持久化

? ? ? ? 3.3 消息持久化

? ? ? ? 3.4 LazyQueue 懒惰队列

? ? ? ? 3.4.1 声明懒惰队列的方式

? ? ? ? 4.0 消费者的可靠性

? ? ? ? 4.1 消费者确认机制

? ? ? ? 4.2 失败重试机制

? ? ? ? 4.3 失败处理策略

? ? ? ? 5.0 使用 DelayExchange 插件

? ? ? ? 5.1 安装 DelayExchange 插件

? ? ? ? 5.2 声明延迟交换机

???5.3 发送延迟消息

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 文件，添加下面的内容：

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

利用命令停掉 RabbitMQ 服务：

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 中添加配置：

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
    publisher-returns: true # 开启publisher return机制

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

1）none：关闭 confirm 机制

2）simple：同步阻塞等待 MQ 的回执

3）correlated：MQ 异步回调返回回执

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

2.2.2 定义 ReturnCallback 机制

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

代码如下：

package com.itheima.publisher.config;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {
    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.error("触发return callback,");
                log.debug("exchange: {}", returned.getExchange());
                log.debug("routingKey: {}", returned.getRoutingKey());
                log.debug("message: {}", returned.getMessage());
                log.debug("replyCode: {}", returned.getReplyCode());
                log.debug("replyText: {}", returned.getReplyText());
            }
        });
    }
}

测试：

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

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

    // 首先引入依赖
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() {
        //发送Object类型的消息
        textObject();
    }

    @Test
    public void textObject(){
        User user = new User(1,"xbs","123456",null,null);
        rabbitTemplate.convertAndSend("textExchange", "xbss", user, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setMessageId("123456");
                return message;
            }
        });
    }

执行结果：

2.2.3 定义 ConfirmCallback 机制

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

代码如下：

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

执行结果：

执行成功之后，日志输出：

接收消息：

注意：

开启生产者确认比较消耗 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 模式：

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

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

@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){
    log.info("接收到 lazy.queue的消息：{}", msg);
}

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 配置如下：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto # 自动ack

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

1）发送消息：

    // 首先引入依赖
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() {
        //发送Object类型的消息
        testSendMessage();
    }
    @Test
    public void testSendMessage(){
        //发送消息
        rabbitTemplate.convertAndSend("yt","xbs","hello rabbitmq");
    }

2）接收消息：

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

执行结果：

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

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

4.2 失败重试机制

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

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

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

修改 application.yml 文件，添加内容：

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

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

结果如下：

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

此时队列中的信息不存在了，丢失了：

结论：

  • 开启本地重试时，消息处理过程中抛出异常，不会 requeue 到队列，而是在消费者本地重试

  • 重试达到最大次数后，Spring 会返回 reject，消息会被丢弃。

4.3 失败处理策略

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

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

  • RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式

  • ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队

  • RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机

比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。

代码实现：

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

@Bean
public DirectExchange errorMessageExchange(){
    return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){
    return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
    return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

2）定义一个 RepublishMessageRecoverer，关联队列和交换机

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

完整代码：

package com.itheima.consumer.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;

@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
public class ErrorMessageConfig {
    @Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue", true);
    }
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }

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

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

测试：

1）发送消息：

    // 首先引入依赖
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() {
        //发送Object类型的消息
        testSendMessage();
    }
    @Test
    public void testSendMessage(){
        //发送消息
        rabbitTemplate.convertAndSend("yt","xbs","hello rabbitmq");
    }

2）接收消息：

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

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

执行结果：

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

5.0 使用 DelayExchange 插件

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

5.1 安装 DelayExchange 插件

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

docker volume inspect mq-plugins

结果如下：

[
    {
        "CreatedAt": "2024-06-19T09:22:59+08:00",
        "Driver": "local",
        "Labels": null,
        "Mountpoint": "/var/lib/docker/volumes/mq-plugins/_data",
        "Name": "mq-plugins",
        "Options": null,
        "Scope": "local"
    }
]

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

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

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

接下来执行命令，安装插件：

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

安装结果：

5.2 声明延迟交换机

1）基于注解方式：

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

2）基于 @Bean 的方式：

package com.itheima.consumer.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Slf4j
@Configuration
public class DelayExchangeConfig {

    @Bean
    public DirectExchange delayExchange(){
        return ExchangeBuilder
                .directExchange("delay.direct") // 指定交换机类型和名称
                .delayed() // 设置delay的属性为true
                .durable(true) // 持久化
                .build();
    }

    @Bean
    public Queue delayedQueue(){
        return new Queue("delay.queue");
    }
    
    @Bean
    public Binding delayQueueBinding(){
        return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
    }
}

5.3 发送延迟消息

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

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

测试：

1）接收消息：

    @Autowired
    MessageConverter messageConverter;

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

2）发送消息：

    // 首先引入依赖
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() {
        //发送Object类型的消息
        testSendDelayMessage();
    }
    
    @Test
    void testSendDelayMessage(){
        //发送消息
        rabbitTemplate.convertAndSend("delay.exchange", "xbs", "hello rabbitmq delay message", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //先设置唯一ID
                message.getMessageProperties().setMessageId("123456");
                //设置延迟发送时间
                message.getMessageProperties().setDelayLong(5000L);
                return message;
            }
        });
    }

执行结果：

注意：

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

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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/924495.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

Redis（概念、IO模型、多路选择算法、安装和启停）

Redis（概念、IO模型、多路选择算法、安装和启停）

一、概念 关系型数据库是典型的行存储数据库，存在的问题是，按行存储的数据在物理层面占用的是连续存储空间，不适合海量数据存储。 Redis在生产中使用的最多的是用作数据缓存。 服务器先在缓存中查询数据，查到则返回，…
阅读更多...
JAVA：Spring Boot 3 实现 Gzip 压缩优化的技术指南

JAVA：Spring Boot 3 实现 Gzip 压缩优化的技术指南

1、简述 随着 Web 应用的用户量和数据量增加，网络带宽和页面加载速度逐渐成为瓶颈。为了减少数据传输量，提高用户体验，我们可以使用 Gzip 压缩 HTTP 响应。本文将介绍如何在 Spring Boot 3 中实现 Gzip 压缩优化。 2、配置 Spring Boot 3 对…
阅读更多...
python期末复习

python期末复习

其他复习资料 Python期末复习-系列数据类型-CSDN博客 期末python复习-异常和函数-CSDN博客 期末Python复习-输入输出-CSDN博客 目录 一、面向对象程序设计 1.思维导图 2.基本概念 3.类对象和实例对象 3.1创建对象 3.2定义类中的成员变量 3.3类中属性的公有和私有 3.…
阅读更多...
HDU Go Running（最小点覆盖 + 网络流优化）

HDU Go Running（最小点覆盖 + 网络流优化）

题目大意：有一条无限长跑道，每个人可以规定自己跑步的方向，起点，跑步起止时间。每个人跑步的速度都是1m/s。最后从监控人员哪里得到了n个报告，每个报告给出了某人在某一时候所在的位置，问跑步的最少可能人数…
阅读更多...
《用Python实现3D动态旋转爱心模型》

《用Python实现3D动态旋转爱心模型》

简介 如果二维的爱心图案已经无法满足你的创意，那今天的内容一定适合你！通过Python和matplotlib库，我们可以实现一个动态旋转的3D爱心模型，充满立体感和动感。# 实现代码（完整代码底部名片私信） 以下是完…
阅读更多...
Unity-Lightmap入门篇

Unity-Lightmap入门篇

：：这是一个实战文章，并没有知识分享，或理论知识；完全没有 关键字： “lightmap","全局光照”，“light Probe" (会混合一些中英文搜索，或者全英文搜索） …
阅读更多...
ElasticSearch通过es-head插件安装可视化及相关问题

ElasticSearch通过es-head插件安装可视化及相关问题

1.es-head下载地址 GitHub - mobz/elasticsearch-head: A web front end for an elastic search cluster 2.启动 建议使用vscode启动，并安装好node.js环境 npm installnpm run start 通过http://localhost:9100就可以看到本地添加的es库 3.相关问题 3.1跨域问…
阅读更多...
Android PMS（Package Manager Service）源码介绍

Android PMS（Package Manager Service）源码介绍

文章目录 前言一、PMS 启动流程二、APK 安装流程三、APK 卸载流程四、权限管理静态权限动态权限 五、 数据存储与一致性六、 PMS 的安全性策略1、权限检查2、签名认证3、动态权限管理4、应用安装验证5、保护系统目录 七、PMS 调试方法总结 前言 PackageManagerService&#xf…
阅读更多...
OSPTrack：一个包含多个生态系统中软件包执行时生成的静态和动态特征的标记数据集，用于识别开源软件中的恶意行为。

OSPTrack：一个包含多个生态系统中软件包执行时生成的静态和动态特征的标记数据集，用于识别开源软件中的恶意行为。

2024-11-22 ，由格拉斯哥大学创建的OSPTrack数据集，目的是通过捕获在隔离环境中执行包和库时生成的特征，包括静态和动态特征，来识别开源软件（OSS）中的恶意指标，特别是在源代码访问受限时&#xf…
阅读更多...
Web登录页面设计

Web登录页面设计

记录第一个前端界面，暑假期间写的，用了Lottie动画和canvas标签做动画，登录和注册也连接了数据库。 图片是从网上找的，如有侵权私信我删除，谢谢啦~
阅读更多...
MySQL45讲 第29讲 如何判断一个数据库是不是出问题了？——阅读总结

MySQL45讲 第29讲 如何判断一个数据库是不是出问题了？——阅读总结

文章目录 MySQL45讲 第二十九讲 如何判断一个数据库是不是出问题了？——阅读总结一、检测数据库实例健康状态的重要性二、常见检测方法及问题分析（一）select 1 判断法（二）查表判断法（三）更新判断…
阅读更多...
mac下Gpt Chrome升级成GptBrowser书签和保存的密码恢复

mac下Gpt Chrome升级成GptBrowser书签和保存的密码恢复

cd /Users/自己的用户名/Library/Application\ Support/ 目录下有 GPT\ Chrome/ Google/ GptBrowser/ GPT\ Chrome 为原来的chrome浏览器的文件存储目录. GptBrowser 为升级后chrome浏览器存储目录 书签所在的文件 Bookmarks 登录账号Login 相关的文件 拷贝到GptBrow…
阅读更多...
论文阅读笔记 | EEG：运动执行过程中的ERD

论文阅读笔记 | EEG：运动执行过程中的ERD

参考：https://mp.weixin.qq.com/s/RmcPSLv1ITMZZwqe2uZ_og?token1093147649&langzh_CN
阅读更多...
Android U ART young cc流程分析

Android U ART young cc流程分析

概述： 众所周知jvm虚拟机为了提高内存回收效率，更高效的进行内存管理与回收，对堆内存进行了分代管理比如hotspot虚拟机的新生代，老年代。根据各代的特征（ 新生代对象分配频繁而生存周期短，老年代生存周期长…
阅读更多...
C++ 11重点总结1

C++ 11重点总结1

智能指针 智能指针: C11引入了四种智能指针: auto_ptr(已弃用)、unique_ptr、shared_ptr和weak_ptr。智能指针可以更有效地管理堆内存,并避免常见的内存泄漏问题。 shared_ptr: 自定义删除器。 shared_ptr使用引用计数来管理它指向的对象的生命周期。多个shared_ptr实例可以指向…
阅读更多...
Sickos1.1 详细靶机思路 实操笔记

Sickos1.1 详细靶机思路 实操笔记

Sickos1.1 详细靶机思路 实操笔记 免责声明 本博客提供的所有信息仅供学习和研究目的，旨在提高读者的网络安全意识和技术能力。请在合法合规的前提下使用本文中提供的任何技术、方法或工具。如果您选择使用本博客中的任何信息进行非法活动，您将独自承担…
阅读更多...
GB28181系列二：SIP信令

GB28181系列二：SIP信令

我的音视频/流媒体开源项目(github) GB28181系列目录 目录 一、SIP报文介绍 二、SIP交互流程： 1、Session Model 2、Pager Model 3、SIP信令交互过程中的3个定义 三、媒体传输（SDP和RTP） 一、SIP报文介绍 这里将会介绍SIP…
阅读更多...
【接口自动化测试】一文从0到1详解接口测试协议！

【接口自动化测试】一文从0到1详解接口测试协议！

接口自动化测试是软件开发过程中重要的环节之一。通过对接口进行测试，可以验证接口的功能和性能，确保系统正常运行。本文将从零开始详细介绍接口测试的协议和规范。 定义接口测试协议 接口测试协议是指用于描述接口测试的规范和约定。它包含了接口的请求…
阅读更多...
CentOS7执行yum命令报错，已加载插件：fastestmirrorLoading mirror speeds from cached hostfile

CentOS7执行yum命令报错，已加载插件：fastestmirrorLoading mirror speeds from cached hostfile

一、出现一下异常问题，表示域名没有配置或配置错误 问题一： 0curl: (6) Could not resolve host: mirrors.aliyun.com; 未知的错误 问题二：虚拟机使用ping主机，提示network unreachable 2.原因分析 出现这个问题是因为yum在安装…
阅读更多...
【Threejs进阶教程-着色器篇】9.顶点着色器入门

【Threejs进阶教程-着色器篇】9.顶点着色器入门

【Threejs进阶教程-着色器篇】9.顶点着色器入门 本系列教程第一篇地址，建议按顺序学习认识顶点着色器varying介绍顶点着色器与片元着色器分别的作用Threejs在Shader中的内置变量各种矩阵gl_Position 尝试使用顶点着色器增加分段数增强效果 制作平面鼓包效果鼓包效果…
阅读更多...
最新文章

Copyright @ 2022~2023