RabbitMQ：生产者可靠性（生产者重连、生产者确认）

---

一、生产者重连

当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试过程中，当前线程是被阻塞的，会影响业务性能。

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

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

二、生产者确认

RabbitMQ有两种确认机制：Publisher Confirm和Publisher Return。开启确认机制后，在MQ成功收到消息后会返回确认消息给生产者，返回的结果有以下几种情况：

1. 消息投递到了MQ，但是路由失败。此时会通过Publisher Return返回路由异常原因，然后返回ACK，告知投递成功。
2. 临时消息投递给了MQ，并且入队成功，返回ACK，告知投递成功。
3. 持久消息投递到MQ，并且入队成功，返回ACK，告知投递成功。
4. 其他情况都会返回NACK，告知投递失败

在生产者服务YML文件中增加一下配置：

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

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

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

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

package com.ming.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error(String.format("ReturnsCallback：消息发送失败，应答码 %s，原因 %s，交换机 %s，路由键 %s，消息 %s ...", returnedMessage.getReplyCode(), returnedMessage.getReplyText(), returnedMessage.getExchange(), returnedMessage.getRoutingKey(), returnedMessage.getMessage()));}});}
}

发送消息，指定消息ID，消息ConfirmCallback，此回调每一次发送消息的时候都要指定：

@Test
public void confirmCallbackTest() throws InterruptedException {// 1. 创建cdCorrelationData cd = new CorrelationData();// 2. 增加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable throwable) {  // Spring消息回调失败log.warn("ConfirmCallback：Spring消息回调失败......", throwable);}@Overridepublic void onSuccess(CorrelationData.Confirm confirm) {  // 消息发送回调if (confirm.isAck()) {log.info("ConfirmCallback：RabbitMQ 消息发送成功，收到ACK......");}else {log.warn("ConfirmCallback：RabbitMQ 消息发送成功，收到NACK......", confirm.getReason());}}});// 3. 发送消息rabbitTemplate.convertAndSend("mt.topic", "china.weather", "黄色警报 ......", cd);  // 成功例子// rabbitTemplate.convertAndSend("mt.topic", "aaa", "黄色警报 ......", cd);  // 失败例子Thread.sleep(2000);
}