0
0

RabbitMQ 高级

1.发送者的可靠性

首先，我们一起分析一下消息丢失的可能性有哪些。消息从发送者发送消息，到消费者处理消息，需要经过的流程是这样的：


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

  • 发送消息时丢失： - 生产者发送消息时连接MQ失败- 生产者发送消息到达MQ后未找到Exchange(交换机)- 生产者发送消息到达MQ的Exchange后，未找到合适的Queue(队列)- 消息到达MQ后，处理消息的进程发生异常
  • MQ导致消息丢失： - 消息到达MQ，保存到队列后，尚未消费就突然宕机
  • 消费者处理消息时： - 消息接收后尚未处理突然宕机- 消息接收后处理过程中抛出异常

综上，我们要解决消息丢失问题，保证MQ的可靠性，就必须从3个方面入手：

  • 确保生产者一定把消息发送到MQ
  • 确保MQ不会将消息弄丢
  • 确保消费者一定要处理消息

1.1.生产者重试机制

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

RabbitTemplate

与MQ连接超时后，多次重试。

修改

publisher

模块的

application.yaml

文件，添加下面的内容：

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

我们利用命令停掉RabbitMQ服务：

docker stop mq

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

1.2.生产者确认机制

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

  • MQ内部处理消息的进程发生了异常
  • 生产者发送消息到达MQ后未找到Exchange
  • 生产者发送消息到达MQ的Exchange后，未找到合适的Queue，因此无法路由

针对上述情况，RabbitMQ提供了生产者消息确认机制，包括

Publisher Confirm

和

Publisher Return

两种。在开启确认机制的情况下，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。

具体如图所示：
image.png
总结如下：

  • 当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ack的确认信息，代表投递成功
  • 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
  • 持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功
  • 其它情况都会返回NACK，告知投递失败

其中

ack

和

nack

属于Publisher Confirm机制，

ack

是投递成功；

nack

是投递失败。而

return

则属于Publisher Return机制。

1.3.实现生产者确认

1.3.1.开启生产者确认

在

publisher

模块的

application.yaml

中添加配置：

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

这里

publisher-confirm-type

有三种模式可选：

  • none：关闭confirm机制
  • simple：同步阻塞等待MQ的回执
  • correlated：MQ异步回调返回回执

一般我们推荐使用

correlated

，回调机制。

1.3.2.定义ReturnCallback

每个

RabbitTemplate

只能配置一个

ReturnCallback

，因此我们可以在配置类中统一设置。我们在publisher模块定义一个配置类：
image.png
内容如下：

packagecom.itheima.publisher.config;importlombok.AllArgsConstructor;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.ReturnedMessage;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.context.annotation.Configuration;importjavax.annotation.PostConstruct;@Slf4j@AllArgsConstructor@ConfigurationpublicclassMqConfig{privatefinalRabbitTemplate rabbitTemplate;@PostConstructpublicvoidinit(){
        rabbitTemplate.setReturnsCallback(newRabbitTemplate.ReturnsCallback(){@OverridepublicvoidreturnedMessage(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());}});}}

1.3.3.定义ConfirmCallback

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

image.png
这里的CorrelationData中包含两个核心的东西：

  • id：消息的唯一标示，MQ对不同的消息的回执以此做判断，避免混淆
  • SettableListenableFuture：回执结果的Future对象

将来MQ的回执就会通过这个

Future

来返回，我们可以提前给

CorrelationData

中的

Future

添加回调函数来处理消息回执：
image.png

我们新建一个测试，向系统自带的交换机发送消息，并且添加

ConfirmCallback

：

@TestvoidtestPublisherConfirm(){// 1.创建CorrelationDataCorrelationData cd =newCorrelationData();// 2.给Future添加ConfirmCallback
    cd.getFuture().addCallback(newListenableFutureCallback<CorrelationData.Confirm>(){@OverridepublicvoidonFailure(Throwable ex){// 2.1.Future发生异常时的处理逻辑，基本不会触发
            log.error("send message fail", ex);}@OverridepublicvoidonSuccess(CorrelationData.Confirm result){// 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容if(result.isAck()){// result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执
                log.debug("发送消息成功，收到 ack!");}else{// result.getReason()，String类型，返回nack时的异常描述
                log.error("发送消息失败，收到 nack, reason : {}", result.getReason());}}});// 3.发送消息
    rabbitTemplate.convertAndSend("hmall.direct","q","hello", cd);}

执行结果如下：
image.png

可以看到，由于传递的

RoutingKey

是错误的，路由失败后，触发了

return callback

，同时也收到了ack。当我们修改为正确的

RoutingKey

以后，就不会触发

return callback

了，只收到ack。
而如果连交换机都是错误的，则只会收到nack。

2.MQ的可靠性

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

2.1.数据持久化

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

  • 交换机持久化
  • 队列持久化
  • 消息持久化

2.1.1.交换机持久化

在控制台的

Exchanges

页面，添加交换机时可以配置交换机的

Durability

参数：
image.png
设置为

Durable

就是持久化模式，

Transient

就是临时模式。

2.1.2.队列持久化

在控制台的Queues页面，添加队列时，同样可以配置队列的

Durability

参数：
image.png

2.1.3.消息持久化

在控制台发送消息的时候，可以添加很多参数，而消息的持久化是要配置一个

properties

：
image.png

2.2.LazyQueue

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

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

一旦出现消息堆积问题，RabbitMQ的内存占用就会越来越高，直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上，这个行为成为

PageOut

. 

PageOut

会耗费一段时间，并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息，生产者的所有请求都会被阻塞。

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

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

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

2.2.1.控制台配置Lazy模式

在添加队列的时候，添加

x-queue-mod=lazy

参数即可设置队列为Lazy模式：
image.png

2.2.2.代码配置Lazy模式

在利用

SpringAMQP

声明队列的时候，添加

x-queue-mod=lazy

参数也可设置队列为Lazy模式：

@BeanpublicQueuelazyQueue(){returnQueueBuilder.durable("lazy.queue").lazy()// 开启Lazy模式.build();}

这里是通过

QueueBuilder

的

lazy()

函数配置Lazy模式，底层源码如下：
image.png

当然，我们也可以基于注解来声明队列并设置为Lazy模式：

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

3.消费者的可靠性

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

  • 消息投递的过程中出现了网络故障
  • 消费者接收到消息后突然宕机
  • 消费者接收到消息后，因处理不当导致异常
  • …

2.1.消费者确认机制

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

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

一般reject方式用的较少，除非是消息格式有问题，那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过

try catch

机制捕获，消息处理成功时返回ack，处理失败时返回nack.

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

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

通过下面的配置可以修改SpringAMQP的ACK处理方式：

spring:rabbitmq:listener:simple:acknowledge-mode: none # 不做处理

修改consumer服务的SpringRabbitListener类中的方法，模拟一个消息处理的异常：

@RabbitListener(queues ="simple.queue")publicvoidlistenSimpleQueueMessage(String msg)throwsInterruptedException{
    log.info("spring 消费者接收到消息：【"+ msg +"】");if(true){thrownewMessageConversionException("故意的");}
    log.info("消息处理完成");}

测试可以发现：当消息处理发生异常时，消息依然被RabbitMQ删除了。

我们再次把确认机制修改为auto：

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

在异常位置打断点，再次发送消息，程序卡在断点时，可以发现此时消息状态为

unacked

（未确定状态）：
image.png
放行以后，由于抛出的是消息转换异常，因此Spring会自动返回

reject

，所以消息依然会被删除：
image.png

我们将异常改为RuntimeException类型：

@RabbitListener(queues ="simple.queue")publicvoidlistenSimpleQueueMessage(String msg)throwsInterruptedException{
    log.info("spring 消费者接收到消息：【"+ msg +"】");if(true){thrownewRuntimeException("故意的");}
    log.info("消息处理完成");}

在异常位置打断点，然后再次发送消息测试，程序卡在断点时，可以发现此时消息状态为

unacked

（未确定状态）：
image.png放行以后，由于抛出的是业务异常，所以Spring返回

ack

，最终消息恢复至

Ready

状态，并且没有被RabbitMQ删除：
image.png
当我们把配置改为

auto

时，消息处理失败后，会回到RabbitMQ，并重新投递到消费者。

2.2.失败重试机制

当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者。如果消费者再次执行依然出错，消息会再次requeue到队列，再次投递，直到消息处理成功为止。极端情况就是消费者一直无法执行成功，那么消息requeue就会无限循环，导致mq的消息处理飙升，带来不必要的压力：

image.png

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

修改consumer服务的application.yml文件，添加内容：

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

重启consumer服务，重复之前的测试。可以发现：

  • 消费者在失败后消息没有重新回到MQ无限重新投递，而是在本地重试了3次
  • 本地重试3次以后，抛出了AmqpRejectAndDontRequeueException异常。查看RabbitMQ控制台，发现消息被删除了，说明最后SpringAMQP返回的是reject

结论：

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

2.3.失败处理策略

在之前的测试中，本地测试达到最大重试次数后，消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下，显然不太合适了。因此Spring允许我们自定义重试次数耗尽后的消息处理策略，这个策略是由

MessageRecovery

接口来定义的，它有3个不同实现：

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

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

RepublishMessageRecoverer

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

1）在consumer服务中定义处理失败消息的交换机和队列

@BeanpublicDirectExchangeerrorMessageExchange(){returnnewDirectExchange("error.direct");}@BeanpublicQueueerrorQueue(){returnnewQueue("error.queue",true);}@BeanpublicBindingerrorBinding(Queue errorQueue,DirectExchange errorMessageExchange){returnBindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}

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

@BeanpublicMessageRecovererrepublishMessageRecoverer(RabbitTemplate rabbitTemplate){returnnewRepublishMessageRecoverer(rabbitTemplate,"error.direct","error");}

完整代码如下：

packagecom.itheima.consumer.config;importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.DirectExchange;importorg.springframework.amqp.core.Queue;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.amqp.rabbit.retry.MessageRecoverer;importorg.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;importorg.springframework.context.annotation.Bean;@Configuration@ConditionalOnProperty(name ="spring.rabbitmq.listener.simple.retry.enabled", havingValue ="true")publicclassErrorMessageConfig{@BeanpublicDirectExchangeerrorMessageExchange(){returnnewDirectExchange("error.direct");}@BeanpublicQueueerrorQueue(){returnnewQueue("error.queue",true);}@BeanpublicBindingerrorBinding(Queue errorQueue,DirectExchange errorMessageExchange){returnBindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}@BeanpublicMessageRecovererrepublishMessageRecoverer(RabbitTemplate rabbitTemplate){returnnewRepublishMessageRecoverer(rabbitTemplate,"error.direct","error");}}

2.4.业务幂等性

幂等性

是一个数学概念，表示函数

f(x) = f(f(x))

。在程序开发中，幂等性指的是同一个业务执行一次或多次对业务状态的影响是一致的。然而，数据的更新往往不是幂等的，重复执行可能导致不同后果。在实际业务场景中，由于意外情况，业务可能会被重复执行，例如页面卡顿、服务间调用重试和MQ消息的重复投递。为了保证消息处理的幂等性，可以采用

唯一消息ID

和

业务状态判断

两种方案。

2.4.1.唯一消息ID

这个思路非常简单：

  1. 每一条消息都生成一个唯一的id，与消息一起投递给消费者。
  2. 消费者接收到消息后处理自己的业务，业务处理成功后将消息ID保存到数据库
  3. 如果下次又收到相同消息，去数据库查询判断是否存在，存在则为重复消息放弃处理。
SpringAMQP

的

MessageConverter

自带了

MessageID

的功能，我们只要开启这个功能即可。以Jackson的消息转换器为例：

@BeanpublicMessageConvertermessageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jjmc =newJackson2JsonMessageConverter();// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jjmc.setCreateMessageIds(true);return jjmc;}

2.4.2.业务判断

业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息，不同的业务场景判断的思路也不一样。相比较而言，消息ID的方案需要改造原有的数据库，所以我更推荐使用业务判断的方案。

以支付修改订单的业务为例，我们需要修改

OrderServiceImpl

中的

markOrderPaySuccess

方法：

@OverridepublicvoidmarkOrderPaySuccess(Long orderId){// 1.查询订单Order old =getById(orderId);// 2.判断订单状态if(old ==null|| old.getStatus()!=1){// 订单不存在或者订单状态不是1，放弃处理return;}// 3.尝试更新订单Order order =newOrder();
        order.setId(orderId);
        order.setStatus(2);
        order.setPayTime(LocalDateTime.now());updateById(order);}

上述代码逻辑上符合了幂等判断的需求，但是由于判断和更新是两步动作，因此在极小概率下可能存在线程安全问题。

我们可以合并上述操作为这样：

@OverridepublicvoidmarkOrderPaySuccess(Long orderId){// UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1lambdaUpdate().set(Order::getStatus,2).set(Order::getPayTime,LocalDateTime.now()).eq(Order::getId, orderId).eq(Order::getStatus,1).update();}

注意看，上述代码等同于这样的SQL语句：

UPDATE`order`SETstatus= ? , pay_time = ? WHERE id = ? ANDstatus=1

我们在where条件中除了判断id以外，还加上了status必须为1的条件。如果条件不符（说明订单已支付），则SQL匹配不到数据，根本不会执行。

4.延迟消息

在电商的支付业务中，对于一些库存有限的商品，为了更好的用户体验，通常都会在用户下单时立刻扣减商品库存。例如电影院购票、高铁购票，下单后就会锁定座位资源，其他人无法重复购买。

但是这样就存在一个问题，假如用户下单后一直不付款，就会一直占有库存资源，导致其他客户无法正常交易，最终导致商户利益受损！

因此，电商中通常的做法就是：对于超过一定时间未支付的订单，应该立刻取消订单并释放占用的库存。但问题来了：如何才能准确的实现在下单后第30分钟去检查支付状态呢？

像这种在一段时间以后才执行的任务，我们称之为延迟任务，而要实现延迟任务，最简单的方案就是利用MQ的延迟消息了。

在RabbitMQ中实现延迟消息也有两种方案：

  • 死信交换机+TTL
  • 延迟消息插件

4.1.死信交换机和延迟消息

4.1.1.死信交换机

什么是死信？当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

  • 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
  • 消息是一个过期消息，超时无人消费
  • 要投递的队列消息满了，无法投递

如果一个队列中的消息已经成为死信，并且这个队列通过

dead-letter-exchange

属性指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为死信交换机（Dead Letter Exchange）。而此时加入有队列与死信交换机绑定，则最终死信就会被投递到这个队列中。

死信交换机有什么作用呢？

  1. 收集那些因处理失败而被拒绝的消息
  2. 收集那些因队列满了而被拒绝的消息
  3. 收集因TTL（有效期）到期的消息 帮我精简这段话

4.1.2.延迟消息

前面两种作用场景可以看做是把死信交换机当做一种消息处理的最终兜底方案，与消费者重试时讲的

RepublishMessageRecoverer

作用类似。

而最后一种场景，大家设想一下这样的场景：如图，有一组绑定的交换机（

ttl.fanout

）和队列（

ttl.queue

）。但是

ttl.queue

没有消费者监听，而是设定了死信交换机

hmall.direct

，而队列

direct.queue1

则与死信交换机绑定，RoutingKey是blue：
image.png

假如我们现在发送一条消息到

ttl.fanout

，RoutingKey为blue，并设置消息的有效期为5000毫秒：
image.png

消息肯定会被投递到

ttl.queue

之后，由于没有消费者，因此消息无人消费。5秒之后，消息的有效期到期，成为死信：
image.png
死信被再次投递到死信交换机

hmall.direct

，并沿用之前的RoutingKey，也就是

blue

：
image.png
由于

direct.queue1

与

hmall.direct

绑定的key是blue，因此最终消息被成功路由到

direct.queue1

，如果此时有消费者与

direct.queue1

绑定， 也就能成功消费消息了。但此时已经是5秒钟以后了：
image.png
也就是说，publisher发送了一条消息，但最终consumer在5秒后才收到消息。我们成功实现了延迟消息。

4.2.DelayExchange插件

基于死信队列虽然可以实现延迟消息，但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果。官方文档说明：Scheduling Messages with RabbitMQ | RabbitMQ - Blog

4.2.1.下载

插件下载地址：
GitHub - rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ
由于我们安装的MQ是

3.8

版本，因此这里下载

3.8.17

版本：
image.png

4.2.2.安装

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

docker volume inspect mq-plugins

结果如下：

[{"CreatedAt":"2023-12-18T14:40:50+08:00",
        "Driver":"local",
        "Labels": null,
        "Mountpoint":"/var/lib/docker/volumes/mq-plugins/_data",
        "Name":"mq-plugins",
        "Options": null,
        "Scope":"local"}]

插件目录被挂载到了

/var/lib/docker/volumes/mq-plugins/_data

这个目录，我们上传插件到该目录下。

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

dockerexec-it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

运行结果如下：

在这里插入图片描述

4.2.3.声明延迟交换机

基于注解方式：

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

基于

@Bean

的方式：

packagecom.itheima.consumer.config;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@Slf4j@ConfigurationpublicclassDelayExchangeConfig{@BeanpublicDirectExchangedelayExchange(){returnExchangeBuilder.directExchange("delay.direct")// 指定交换机类型和名称.delayed()// 设置delay的属性为true.durable(true)// 持久化.build();}@BeanpublicQueuedelayedQueue(){returnnewQueue("delay.queue");}@BeanpublicBindingdelayQueueBinding(){returnBindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");}}

4.2.4.发送延迟消息

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

@TestvoidtestPublisherDelayMessage(){// 1.创建消息String message ="hello, delayed message";// 2.发送消息，利用消息后置处理器添加消息头
    rabbitTemplate.convertAndSend("delay.direct","delay", message,newMessagePostProcessor(){@OverridepublicMessagepostProcessMessage(Message message)throwsAmqpException{// 添加延迟消息属性
            message.getMessageProperties().setDelay(5000);return message;}});}
标签： rabbitmq ruby 中间件
本文转载自: https://blog.csdn.net/weixin_46370595/article/details/135084734
版权归原作者 小钟不想敲代码 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ 高级”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航