弹性：从错误和 Broker 故障中恢复
Spring AMQP 提供的一些关键（也是最受欢迎的）高级功能与协议错误或 broker 故障发生时的恢复和自动重连有关。本指南中我们已经见过所有相关组件，但在这里把它们汇集到一起，并逐一说明这些功能和恢复场景会有所帮助。
主要的重连功能由 CachingConnectionFactory 本身启用。使用 RabbitAdmin 的自动声明功能通常也很有益。此外，如果你关心可靠投递，可能还需要在 RabbitTemplate 和 SimpleMessageListenerContainer 中使用 channelTransacted 标志，并在 SimpleMessageListenerContainer 中使用 AcknowledgeMode.AUTO（如果你自己处理确认，则用手动模式）。
自动声明交换机、队列和绑定
RabbitAdmin 组件可以在启动时声明交换机、队列和绑定。它通过 ConnectionListener 惰性地完成此操作。因此，如果启动时 broker 不存在，也没有关系。第一次使用 Connection 时（例如发送消息时），监听器触发并应用 admin 功能。在监听器中执行自动声明的另一个好处是，如果连接因任何原因断开（例如 broker 宕机、网络故障等），在连接重新建立时会再次应用这些声明。
以这种方式声明的队列必须具有固定名称——要么显式声明，要么由框架为 AnonymousQueue 实例生成。匿名队列是非持久、独占且自动删除的。
自动声明仅在 CachingConnectionFactory 缓存模式为 CHANNEL（默认）时执行。存在此限制是因为独占和自动删除队列绑定到连接。
从 2.2.2 版本开始，RabbitAdmin 会检测类型为 DeclarableCustomizer 的 bean，并在实际处理声明之前应用该函数。这很有用，例如在某个参数获得框架一等支持之前设置它的新参数（属性）。
@Bean
public DeclarableCustomizer customizer() {
return dec -> {
if (dec instanceof Queue && ((Queue) dec).getName().equals("my.queue")) {
dec.addArgument("some.new.queue.argument", true);
}
return dec;
};
}
这在不提供对 Declarable bean 定义直接访问的项目中也很有用。
另请参见 RabbitMQ 自动连接/拓扑恢复。
同步操作中的失败与重试选项
如果在使用 RabbitTemplate 时某个同步序列中丢失了与 broker 的连接，Spring AMQP 会抛出 AmqpException（通常是 AmqpIOException，但不总是）。我们不会试图掩盖存在问题这一事实，因此你必须能够捕获并响应该异常。如果你怀疑连接已丢失（且不是你的过错），最简单的做法是重试该操作。你可以手动重试，也可以考虑使用 Spring Retry 来处理重试（命令式或声明式）。
Spring Retry 提供了几个 AOP 拦截器和极大的灵活性来指定重试参数（重试次数、异常类型、退避算法等）。Spring AMQP 还提供了一些便捷的工厂 bean，以方便的形式为 AMQP 用例创建 Spring Retry 拦截器，并带有强类型回调接口，你可以用它们实现自定义恢复逻辑。更多细节参见 StatefulRetryOperationsInterceptorFactoryBean 和 StatelessRetryOperationsInterceptorFactoryBean 的 Javadoc 和属性。如果没有事务，或者事务在重试回调内部启动，则适合使用无状态重试。注意，无状态重试比有状态重试更易于配置和分析，但如果存在一个必须回滚或必定会回滚的进行中事务，它通常并不合适。事务进行中连接断开应与回滚具有相同效果。因此，对于在栈更上层启动事务的重连场景，有状态重试通常是最佳选择。有状态重试需要一个唯一标识消息的机制。最简单的方式是让发送方在 MessageId 消息属性中放入一个唯一值。提供的消息转换器提供了这样做的选项：你可以把 createMessageIds 设置为 true。否则，你可以向拦截器注入一个 MessageKeyGenerator 实现。键生成器必须为每条消息返回唯一键。在 2.0 之前的版本中，提供了一个 MissingMessageIdAdvice。它让没有 messageId 属性的消息恰好重试一次（忽略重试设置）。此 advice 不再提供，因为从 spring-retry 1.2 开始，其功能已内置于拦截器和消息监听器容器中。
为向后兼容，消息 ID 为 null 的消息默认被视为对消费者致命（消费者停止）（在一次重试之后）。要复制 MissingMessageIdAdvice 提供的功能，你可以在监听器容器上把 statefulRetryFatalWithNullMessageId 属性设置为 false。设置后，消费者继续运行，消息被拒绝（在一次重试之后）。它会被丢弃或路由到死信队列（如果配置了）。
从 1.3 版本开始，提供了一个构建器 API，以辅助用 Java（在 @Configuration 类中）组装这些拦截器。下例展示了如何这样做：
@Bean
public StatefulRetryOperationsInterceptor interceptor() {
return RetryInterceptorBuilder.stateful()
.maxRetries(5)
.backOffOptions(1000, 2.0, 10000) // initialInterval, multiplier, maxInterval
.build();
}
只有一部分重试功能可以这样配置。更高级的功能需要配置 RetryPolicy。关于可用策略及其配置的更多信息，参见 RetryPolicy Javadoc。
批量监听器的重试
不建议为批量监听器配置重试，除非批次是由生产者在单条记录中创建的。关于消费者和生产者创建的批次的信息，参见批量消息。对于消费者创建的批次，框架不知道批次中哪条消息导致了失败，因此重试耗尽后无法恢复。对于生产者创建的批次，由于只有一条消息真正失败，整条消息都可以被恢复。应用程序可能希望告知自定义恢复器失败发生在批次中的位置，例如通过在抛出的异常上设置一个索引属性。
批量监听器的重试恢复器必须实现 MessageBatchRecoverer。
消息监听器与异步情形
如果 MessageListener 因业务异常失败，该异常由消息监听器容器处理，容器随后回去继续监听下一条消息。如果失败是由连接断开（而非业务异常）引起的，则必须取消并重新启动为监听器收集消息的消费者。SimpleMessageListenerContainer 无缝处理这一点，并记录日志说明监听器正在重启。事实上，它会无休止地循环，尝试重启消费者。只有当消费者确实表现得非常糟糕时它才会放弃。一个副作用是：如果容器启动时 broker 宕机，它会持续尝试，直到建立连接。
与协议错误和连接断开不同，业务异常处理可能需要更多考虑和一些自定义配置，尤其是在使用事务或容器确认时。在 2.8.x 之前，RabbitMQ 没有定义死信行为。因此，默认情况下，因业务异常被拒绝或回滚的消息可能被无休止地重新投递。要限制客户端的重新投递次数，一个选择是在监听器的通知链中使用 StatefulRetryOperationsInterceptor。该拦截器可以有一个恢复回调，实现自定义死信操作——无论什么操作适合你的特定环境。
另一种选择是把容器的 defaultRequeueRejected 属性设置为 false。这会导致所有失败消息被丢弃。使用 RabbitMQ 2.8.x 或更高版本时，这也便于把消息投递到死信交换机。
或者，你可以抛出 AmqpRejectAndDontRequeueException。这样做会阻止消息重新入队，无论 defaultRequeueRejected 属性如何设置。
从 2.1 版本开始，引入了 ImmediateRequeueAmqpException，执行完全相反的逻辑：消息会被重新入队，无论 defaultRequeueRejected 属性如何设置。
通常会结合使用这两种技术。你可以在通知链中使用 StatefulRetryOperationsInterceptor，并配合一个抛出 AmqpRejectAndDontRequeueException 的 MessageRecoverer。当所有重试都耗尽时会调用 MessageRecover。RejectAndDontRequeueRecoverer 正是这样做的。默认的 MessageRecoverer 会消费这条出错消息并发出一条 WARN 消息。
从 1.3 版本开始，提供了一个新的 RepublishMessageRecoverer，允许在重试耗尽后发布失败消息。
当恢复器消费最终异常时，消息会被确认，broker 不会把它发送到死信交换机（如果配置了）。
在消费者一侧使用 RepublishMessageRecoverer 时，收到的消息在 receivedDeliveryMode 消息属性中带有 deliveryMode。在这种情况下，deliveryMode 为 null。这表示 broker 上的 NON_PERSISTENT 投递模式。从 2.0 版本开始，你可以为 RepublishMessageRecoverer 配置 deliveryMode，以便在其为 null 时设置到要重新发布的消息中。默认使用 MessageProperties 的默认值——MessageDeliveryMode.PERSISTENT。
下例展示了如何设置 RepublishMessageRecoverer 作为恢复器：
@Bean
RetryOperationsInterceptor interceptor() {
return RetryInterceptorBuilder.stateless()
.maxRetries(5)
.recoverer(new RepublishMessageRecoverer(amqpTemplate(), "something", "somethingelse"))
.build();
}
RepublishMessageRecoverer 会在消息头中携带额外信息来发布消息，例如异常消息、原始交换机和路由键。出于堆栈大小或敏感信息的考虑，默认不包含错误堆栈。应使用 includeStackTrace(true) 选项来包含专用的堆栈跟踪头。可以通过创建子类并重写 additionalHeaders() 来添加额外的头。也可以在 additionalHeaders() 中更改 deliveryMode（或任何其他属性），如下例所示：
RepublishMessageRecoverer recoverer = new RepublishMessageRecoverer(amqpTemplate, "error") {
protected Map<? extends String, ? extends Object> additionalHeaders(Message message, Throwable cause) {
message.getMessageProperties()
.setDeliveryMode(message.getMessageProperties().getReceivedDeliveryMode());
return null;
}
};
从 2.0.5 版本开始，如果堆栈太大可能会被截断；这是因为所有头必须放在单个帧中。默认情况下，如果堆栈跟踪会导致其他头只剩不到 20,000 字节（“headroom”）可用，它会被截断。如果你需要为其他头预留更多或更少空间，可以通过设置恢复器的 frameMaxHeadroom 属性来调整。从 2.1.13、2.2.3 版本开始，异常消息也计入此计算，堆栈跟踪的量将按以下算法最大化：
- 如果仅堆栈跟踪就会超过限制，异常消息头将被截断为 97 字节加
…，堆栈跟踪也会被截断。 - 如果堆栈跟踪很小，则消息将被截断（加上
…）以适应可用字节（但堆栈跟踪内部的消息本身会被截断为 97 字节加…）。
每当发生任何形式的截断时，原始异常都会被记录日志，以保留完整信息。该评估在头增强之后执行，因此表达式中可以使用异常类型等信息。
从 2.4.8 版本开始，错误交换机和路由键可以以 SpEL 表达式提供，Message 作为求值的根对象。
从 2.3.3 版本开始，提供了一个新子类 RepublishMessageRecovererWithConfirms；它同时支持两种风格的发布确认，并会在返回前等待确认（如果未确认或消息被退回，则抛出异常）。
如果确认类型为 CORRELATED，该子类还会检测消息是否被退回并抛出 AmqpMessageReturnedException；如果发布被否定确认，它会抛出 AmqpNackReceivedException。
如果确认类型为 SIMPLE，该子类会在 channel 上调用 waitForConfirmsOrDie 方法。
关于确认和退回的更多信息，参见发布确认与退回。
从 2.1 版本开始，新增了一个 ImmediateRequeueMessageRecoverer，用于抛出 ImmediateRequeueAmqpException，通知监听器容器重新入队当前失败消息。
Spring Retry 的异常分类
Spring Retry 在确定哪些异常可以触发重试方面具有极大的灵活性。默认配置对所有异常都重试。鉴于用户异常被包装在 ListenerExecutionFailedException 中，我们需要确保分类会检查异常的原因。默认分类器只查看顶层异常。
从 Spring Retry 1.0.3 开始，BinaryExceptionClassifier 有一个名为 traverseCauses 的属性（默认值：false）。当为 true 时，它会遍历异常原因，直到找到匹配项或没有原因为止。
要把此分类器用于重试，你可以使用一个 SimpleRetryPolicy，它由接收最大重试次数、Exception 实例的 Map 和布尔值（traverseCauses）的构造函数创建，并把该策略注入 RetryTemplate。
通过 Broker 重试
从队列进入死信的消息，可以在经 DLX 重新路由后重新发布回该队列。这种重试行为在 broker 一侧通过 x-death 头控制。关于这种方式的更多信息，参见 RabbitMQ 官方文档。
另一种方式是从应用程序手动把失败消息重新发布回原始交换机。从 4.0 版本开始，RabbitMQ broker 不再考虑客户端发送的 x-death 头。本质上，来自客户端的任何 x-* 头都会被忽略。
为缓解 RabbitMQ broker 的这种新行为，Spring AMQP 从 3.2 版本开始引入了 retry_count 头。当此头不存在且服务器端 DLX 生效时，x-death.count 属性会被映射到此头。手动重新发布失败消息进行重试时，必须手动递增 retry_count 头的值。更多信息参见 Javadoc。
下例总结了通过 broker 手动重试的算法：
@RabbitListener(queues = "some_queue")
public void rePublish(Message message) {
try {
// Process message
}
catch (Exception ex) {
Long retryCount = message.getMessageProperties().getRetryCount();
if (retryCount < 3) {
message.getMessageProperties().incrementRetryCount();
this.rabbitTemplate.send("", "some_queue", message);
}
else {
throw new ImmediateAcknowledgeAmqpException("Failed after 4 attempts");
}
}
}