RabbitMQ入门指南(七)：生产者可靠性

专栏导航

RabbitMQ入门指南

从零开始了解大数据

专栏导航

前言

一、消息丢失的可能性

1.发送消息时丢失：

2.MQ导致消息丢失：

3.消费者处理消息时消息丢失：

二、生产者可靠性

1.生产者重试机制

2.生产者确认机制

总结

前言

RabbitMQ是一个高效、可靠的开源消息队列系统，广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了消息丢失的可能性、生产者可靠性中的生产者重试机制和生产者确认机制等内容。

一、消息丢失的可能性

流程图如下：

1.发送消息时丢失：

生产者发送消息时连接MQ失败：当生产者尝试与RabbitMQ建立连接但失败时，发送消息的操作将无法完成。这可能是由于网络问题、RabbitMQ服务未运行或配置错误导致的。
生产者发送消息到达MQ后未找到Exchange：在RabbitMQ中，生产者将消息发送到Exchange，Exchange再路由到一个或多个Queue。如果生产者发送消息到达MQ后未找到正确的Exchange，那么消息将无法被路由到正确的Queue，从而丢失。
生产者发送消息到达MQ的Exchange后，未找到合适的Queue：即使消息找到了Exchange，如果Exchange没有正确配置或没有找到合适的Queue来路由消息，那么消息也会丢失。这可能是由于配置错误或Queue不存在导致的。
消息到达MQ后，处理消息的进程发生异常：在某些情况下，当消息到达MQ后，处理消息的进程可能会发生异常，导致消息无法被处理。这可能是由于代码错误、资源限制或其他系统问题导致的。

2.MQ导致消息丢失：

消息到达MQ，保存到队列后，尚未消费就突然宕机：在某些情况下，如果RabbitMQ服务突然宕机，而消息已经保存到队列中但尚未被消费者消费，那么这些消息可能会丢失。为了防止这种情况，可以使用持久化配置来确保消息在MQ宕机后仍然保留在磁盘上。

3.消费者处理消息时消息丢失：

消息接收后尚未处理突然宕机：如果消费者在接收消息后尚未处理就突然宕机，那么这些消息可能会丢失。为了防止这种情况，可以使用确认机制来确保消息在被消费者成功处理之前一直保留在队列中。
消息接收后处理过程中抛出异常：在处理消息的过程中，如果消费者代码出现异常并抛出错误，那么可能会导致消息处理失败。为了防止这种情况，可以对消费者的代码进行充分的测试和异常处理，以确保其能够正确、稳定地处理消息。

二、生产者可靠性

1.生产者重试机制

由于网络波动，可能会出现客户端连接MO失败的情况，为了解决这个问题，Spring AMQP提供了消息发送时的重试机制。

在publisher服务的application.yml文件中添加配置：

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

当网络不稳定时，重试机制可以有效提高消息发送的成功率。然而，Spring AMQP提供的重试机制是阻塞式的，这意味着在等待重试的过程中，当前线程会被阻塞。对于对业务性能有要求的应用，建议禁用重试机制。如果仍需使用，建议合理配置等待时长和重试次数，并考虑使用异步线程来执行发送消息的代码，以避免阻塞主线程。

2.生产者确认机制

RabbitMQ提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。返回的结果有以下几种情况：

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

ACK和NACK属于Publisher Confirm机制，ACK是投递成功，NACK是投递失败。return属于Publisher Return机制。默认两种机制都是关闭状态，需要通过配置文件来开启。

开启生产者确认：

在publisher服务的application.yml文件中添加配置：

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

publisher-confirm-type有三种模式（一般推荐使用correlated回调机制）：

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

定义ReturnCallback：

在publisher服务定义一个配置类（每个RabbitTemplate只能配置一个ReturnCallback）：

@Slf4j
@Configuration
public class MqConfirmConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 配置回调rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returned) {log.info("消息的return callback:");log.info("exchange:{}, key:{}, msg:{}, code:{}, text:{}",returned.getExchange(), returned.getRoutingKey(), returned.getMessage(), returned.getReplyCode(), returned.getReplyText());}});}
}

定义ConfirmCallback：

由于每条消息的处理逻辑可能不同，因此需要在每次发送消息时定义ConfirmCallback（当调用RabbitTemplate的convertAndSend方法时，需要多传递一个参数）。

这个参数是一个CorrelationData对象，包含两个核心内容：

id：这是消息的唯一标识，MQ会使用它来判断不同的消息的回执，避免混淆。
SettableListenableFuture：这是一个Future对象，MQ将通过它返回回执结果。可以提前为CorrelationData中的Future添加回调函数来处理消息回执。

在publisher服务中的测试类添加一个测试方法，实现消息发送，并且添加ConfirmCallback：

    @Testvoid testConfirmCallback() throws InterruptedException {// 1.创建CorrelationDataCorrelationData cd = new CorrelationData(UUID.randomUUID().toString());// 2.添加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable ex) {log.error("消息回调失败", ex);}@Overridepublic void onSuccess(CorrelationData.Confirm result) {log.debug("收到confirm callback回执");if (result.isAck()) {// 消息发送成功log.debug("消息发送成功，收到ACK");} else {// 消息发送失败log.error("消息发送失败，收到NACK，原因：{}", result.getReason());}}});rabbitTemplate.convertAndSend("test.direct", "red", "test", cd);}

生产者确认会带来额外的网络和系统资源开销，因此应尽量避免使用。如果确实需要使用，则无需开启Publisher-Return机制，因为一般路由失败是业务问题。对于NACK消息，可以有限次数重试，如果仍然失败，则记录异常消息。