RabbitMQ提升消息传输可靠性方法总结

一、简介

在RabbitMQ消息传输过程中，由于网络、服务器本身等问题，有时候会发生消息丢失的现象，在生产环境中，我们应该尽可能保证消息不会丢失，当然不是说百分百保证。下面列举一些发生消息丢失的场景：

【a】如果在声明交换机、队列、发送消息的时候没有指定消息持久化的话，那么在RabbitMQ服务器重启的情况下，所有未持久化的交换机、队列、消息都将会丢失；
【b】如果生产者发送完消息之后，如果在发送到MQ服务器的过程中，发生了异常情况，那么消息将不能在正确到达MQ服务器，此时消息将会发生丢失；
【c】如果发送到交换机的消息没有找到匹配的队列，此时队列中的消息将会发生丢失；
【d】如果我们采用RabbitMQ自动确认机制的话，只要生产者发送完消息，就会认为消费者已经成功处理了，而实际情况是，在消费者消费过程可能发生异常情况，这时候如果自动确认，那么未被消费者消费的消息将会发生丢失；

以上总结的四点，都是会影响消息可靠性传输的一些因素，下面针对上面提到的四点分别总结一下如何提供消息可靠性。

二、方法总结

【a】如果在声明交换机、队列、发送消息的时候没有指定消息持久化的话，那么在RabbitMQ服务器重启的情况下，所有未持久化的交换机、队列、消息都将会丢失；

针对这种情况，我们可以对交换机、队列、已经在发送消息的时候，指定durable=true,将交换机、队列、消息都进行持久化，这样在MQ重启之后，队列中的消息依旧能够被消费者继续消费。

交换机持久化：声明交换机的时候指定参数durable=true，实现交换机持久化。如果交换器不设置持久化，那么在RabbitMQ服务器重启之后，相关的交换器元数据会丢失，消息不会丢失，只是不能将消息发送到这个交换器中。

public TopicExchange(String name, boolean durable, boolean autoDelete) {super(name, durable, autoDelete);}public AbstractExchange(String name) {this(name, true, false);}public AbstractExchange(String name, boolean durable, boolean autoDelete) {this(name, durable, autoDelete, (Map)null);}

new TopicExchange(DURABLE_EXCHANGE_NAME,true,false)

队列持久化：声明交换机的时候指定参数durable=true，实现队列持久化，队列的持久化不能保证内存存储的消息不会丢失。

       public Queue(String name) {this(name, true, false, false);}public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete)//不指定durable的话默认好像也是true//public Queue(String name, boolean durable)//durable：是否将队列持久化 true表示需要持久化 false表示不需要持久化public Queue(String name, boolean durable) {this(name, durable, false, false, (Map)null);}

```
new Queue(DURABLE_QUEUE_NAME, true);
```
消息持久化：

* 使用convertAndSend方式发送消息，消息默认就是持久化的，下面是源码：* new MessageProperties() --> DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT --> deliveryMode = 2;

只有实现了交换机、队列与消息的持久化，才能保证消息不会丢失。

【b】如果生产者发送完消息之后，如果在发送到MQ服务器的过程中，发生了异常情况，那么消息将不能在正确到达MQ服务器，此时消息将会发生丢失；

针对这种情况，我们可以使用生产者发送消息确认机制，每当生产者发送消息到MQ服务时，会监听一个confirmListener，触发下面的方法，生产者可以根据返回的状态进行后续操作。

/*** 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;* 如果消息正确到达交换机,则该方法中isSendSuccess = true;*/@Overridepublic void confirm(CorrelationData correlationData, boolean isSendSuccess, String error) {logger.info("confirm回调方法>>>回调消息ID为: " + correlationData.getId());if (isSendSuccess) {logger.info("confirm回调方法>>>消息发送到交换机成功！");} else {logger.info("confirm回调方法>>>消息发送到交换机失败！，原因 : [{}]", error);}}

实现方法示例如下：实现ConfirmCallback接口,实现： public void confirm(CorrelationData correlationData, boolean isSendSuccess, String error)方法。

/*** @Description 自定义消息发送确认的回调* @Author weishihuai* @Date 2019/6/27 10:42* <p>* 实现接口：implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback* ConfirmCallback：只确认消息是否正确到达交换机中,不管是否到达交换机,该回调都会执行;* ReturnCallback：如果消息从交换机未正确到达队列中将会执行，正确到达则不执行;*/
@Component
public class CustomConfirmAndReturnCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {private static final Logger logger = LoggerFactory.getLogger(CustomConfirmAndReturnCallback.class);@Autowiredprivate RabbitTemplate rabbitTemplate;/*** PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化.*/@PostConstructpublic void init() {//指定 ConfirmCallbackrabbitTemplate.setConfirmCallback(this);//指定 ReturnCallbackrabbitTemplate.setReturnCallback(this);}/*** 消息从交换机成功到达队列，则returnedMessage方法不会执行;* 消息从交换机未能成功到达队列，则returnedMessage方法会执行;*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {logger.info("returnedMessage回调方法>>>" + new String(message.getBody(), StandardCharsets.UTF_8) + ",replyCode:" + replyCode+ ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);}/*** 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;* 如果消息正确到达交换机,则该方法中isSendSuccess = true;*/@Overridepublic void confirm(CorrelationData correlationData, boolean isSendSuccess, String error) {logger.info("confirm回调方法>>>回调消息ID为: " + correlationData.getId());if (isSendSuccess) {logger.info("confirm回调方法>>>消息发送到交换机成功！");} else {logger.info("confirm回调方法>>>消息发送到交换机失败！，原因 : [{}]", error);}}}

当然除了这种生产者confirm确认机制外，还要一种事务的方式，通过channel.txSelect() / channel.txCommit() / channel.txRollback()来控制事务的提交和回滚操作。生产者确认机制相对于事务机制，最大的好处就是可以异步处理提高吞吐量，不需要额外等待消耗资源。实际生产环境推荐使用生产者确认机制。

【c】如果发送到交换机的消息没有找到匹配的队列，此时队列中的消息将会发生丢失；

针对这种未被正确路由的消息，主要有两种方法：

第一种方法：可以通过监听RabbitTemplate.ReturnCallback返回回调spring.rabbitmq.publisher-returns=true，同时设置spring.rabbitmq.template.mandatory = true,这样的话就算消息没有匹配到合适的队列，RabbitMQ也会调用将消息返回给生产者，消息将不会丢失。

实现方法示例：实现RabbitTemplate.ReturnCallback接口，实现
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {}方法

  /*** 消息从交换机成功到达队列，则returnedMessage方法不会执行;* 消息从交换机未能成功到达队列，则returnedMessage方法会执行;*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {logger.info("returnedMessage回调方法>>>" + new String(message.getBody(), StandardCharsets.UTF_8) + ",replyCode:" + replyCode+ ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);}

.第二种方法：通过指定备份交换机实现：在声明交换机的时候添加alternate-exchange参数，将没有被路由的消息存储于RabbitMQ中。

 @Beanpublic Exchange exchange() {Map<String, Object> arguments = new HashMap<>(10);//声明备份交换机arguments.put("alternate-exchange", MESSAGE_BAK_EXCHANGE_NAME);return new DirectExchange(EXCHANGE_NAME, true, false, arguments);}

arguments.put("alternate-exchange", MESSAGE_BAK_EXCHANGE_NAME);

【d】如果我们采用RabbitMQ自动确认机制的话，只要生产者发送完消息，就会认为消费者已经成功处理了，而实际情况是，在消费者消费过程可能发生异常情况，这时候如果自动确认，那么未被消费者消费的消息将会发生丢失；

针对这种情况，我们可以采用消费者手动确认消息机制，即basicAck = false,只有rabbitmq接收到消费者发送的ack应答之后，才会发送下一条消息到该消费者，才会将消息从内存中移除。强烈推荐使用手动确认机制，因为使用手动确认有足够的时间处理消息，不需要担心消费者进程挂掉之后消息丢失问题。

RabbitMQ如果一直没收到消费者的ack消息应答确认，并且此时消费者连接已断开，那么MQ会重新将消息进入队列等待发送给下一个消费者进行消费。

实现方式示例：

@Component
public class Consumer {private static final Logger logger = LoggerFactory.getLogger(Consumer.class);@RabbitListener(queues = "test_dlx_queue_name")public void receiveMessage(String receiveMessage, Message message, Channel channel) {try {logger.info("【Consumer】接收到消息:[{}]", receiveMessage);//这里模拟随机拒绝一些消息到死信队列中if (new Random().nextInt(10) < 5) {logger.info("【Consumer】拒绝一条信息:[{}]，该消息将会被转发到死信交换器中", receiveMessage);channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);} else {channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}} catch (Exception e) {logger.info("【Consumer】接消息后的处理发生异常", e);try {channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (IOException e1) {logger.error("手动确认消息异常.", e1);}}}}

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

三、总结

以上就是关于在RabbitMQ中，对如何提高消息传输可靠性进行了总结，当然并没有哪一种方式能够百分百保证消息可靠性传输，我们只能够尽可能的保证消息不会发生丢失。其实，总结的以上四种方法，在前面的博客中都分别进行了讲解，大家可能选择性的进行阅读总结，希望能对大家有所帮助。