IT评测·应用市场-qidao123.com

标题: 【Rabbitmq】Rabbitmq高级特性-发送者可靠性 [打印本页]

---

作者: 兜兜零元    时间: 2025-1-23 02:39
标题: 【Rabbitmq】Rabbitmq高级特性-发送者可靠性
Rabbitmq提供了两种发送来保证发送者的可靠性，第一种叫发送者重连，第二种叫发送者确认。
发送者重连

有时候由于网络波动，大概会出现发送者毗连MQ失败的环境，通过配置可以开启毗连失败后的重连机制：

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 #最大重试次数

复制代码

注：
当网络不稳定的时候，利用重试机制可以有效进步消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当火线程是被阻塞的，会影响业务性能。
如果对于业务性能有要求，发起禁用重试机制。如果一定要利用，请合理配置等待时长和重试次数，当然也可以考虑利用异步线程来实验发送消息的代码
发送者确认

SpringAMQP提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后，当发送者发送消息给MQ后，MQ会返回确认效果给发送者。返回的效果有以下几种环境：

• 消息投递到了MQ，但是路由失败。此时会通过PublisherReturn返回路由异常原因，然后返回ACK，告知投递成功
  
• 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
  
• 持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功
  
• 别的环境都会返回NACK，告知投递失败
  

1.开启确认机制

在publisher这个微服务的application.yml中添加配置：

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

复制代码

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

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

2.ReturnCallback

每个RabbitTemplate只能配置一个ReturnCallback，因此必要在项目启动过程中配置：

2. @Slf4j
3. @Configuration
4. @RequiredArgsConstructor
5. public class MqConfig {
7.     private final RabbitTemplate rabbitTemplate;
9.     @PostConstruct
10.     public void init(){
11.         rabbitTemplate.setReturnsCallback(returned -> {
12.             //业务处理
13.             log.error("触发return callback,");
14.             log.debug("exchange: {}", returned.getExchange());
15.             log.debug("routingKey: {}", returned.getRoutingKey());
16.             log.debug("message: {}", returned.getMessage());
17.             log.debug("replyCode: {}", returned.getReplyCode());
18.             log.debug("replyText: {}", returned.getReplyText());
19.         });
20.     }
21. }

复制代码

3.ConfirmCallback

发送消息，指定消息ID、消息ConfirmCallback

1.   @Test
2.     public void SendConfirmCallBack() throws InterruptedException {
3.         // 1.创建CorrelationData
4.         CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
5.         // 2.给Future添加ConfirmCallback
6.         cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
7.             @Override
8.             public void onFailure(Throwable ex) {
9.                 // 2.1.Future发生异常时的处理逻辑，基本不会触发
10.                 log.error("handle message ack fail", ex);
11.             }
13.             @Override
14.             public void onSuccess(CorrelationData.Confirm result) {
15.                 // 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容
16.                 if (result.isAck()){// result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执
17.                     log.error("发送消息成功，收到 ack!");
18.                 }else {
19.                     // result.getReason()，String类型，返回nack时的异常描述
20.                     log.error("发送消息失败，收到 nack, reason : {}", result.getReason());
21.                 }
23.             }
24.         });
25.         String exchangeName = "amq.topic";
26.         String message = "Hello topic china";
27.         this.rabbitTemplate.convertAndSend(exchangeName,".11news11", message,cd);
29.         Thread.sleep(3000); // 测试方法用于回显日志
30.     }

复制代码

注：
发送者确认机制必要跟MQ进行通讯和确认会影响发送的效率，开启根据实际需求考虑，开以后要留意重试次数不要无穷重试，留意重试次数，否则对性能影像严重。
MQ的可靠性

在默认环境下，RabbitMQ会将接收到的信息生存在内存中以降低消息收发的耽误。这样会导致两个问题：

• 一旦MQ宕机，内存中的消息会丢失
  
• 内存空间有限，当消耗者故障或处理过慢时，会导致消息积压，引发MQ阻塞
  

数据持久化

RabbitMQ实现数据持久化包罗3个方面：
交换机持久化

交换机默认持久化
Durable 持久
Transient 临时

队列持久化

队列默认持久化
Durable 持久
Transient 临时

消息持久化

消息持久化必要手动指定
Non-persistent 非持久化
Persistent 持久化

代码实现

1. MessageBuilder.withBody("案例".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();

复制代码

注:
由于Mq默认持久化,以是一般不必要修改
Lazy Queue

从RabbitMQ的3.6.0版本开始，就增加了Lazy Queue的概念，也就是惰性队列。
惰性队列的特征如下：

• 接收到消息后直接存入磁盘，不再存储到内存
  
• 消耗者要消耗消息时才会从磁盘中读取并加载到内存（可以提前缓存部分消息到内存，最多2048条）
  在3.12版本后，全队伍列都是Lazy Queue模式，无法更改。
  

要设置一个队列为惰性队列，只必要在声明队列时，指定x-queue-mode属性为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.     }

复制代码

声明式

1.     @Bean
2.     public Queue lazyQueue(){
3.         return QueueBuilder
4.                 .durable("lazy.queue")
5.                 .lazy() // 开启Lazy模式            
6.                 .build();
7.     }

复制代码

总结

• 起首通过配置可以让交换机、队列、以及发送的消息都持久化。这样队列中的消息会持久化到磁盘，MQ重启消息依然存在。
  
• RabbitMQ在3.6版本引入了LazyQueue，并且在3.12版本后会称为队列的默认模式。LazyQueue会将全部消息都持久化
  
• 开启持久化和发送者确认时， RabbitMQ只有在消息持久化完成后才会给发送者返回ACK回执
  

其他文章

发送者可靠性
消耗者可靠性
耽误信息

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/)

Powered by Discuz! X3.4