错误处理
在本节中，我们将解释框架提供的错误处理机制背后的总体思路。我们将以 Rabbit 绑定器为例，因为各个绑定器为某些支持的机制定义了不同的属性集，这些机制特定于底层代理的能力（例如 Kafka 绑定器）。
错误时有发生，Spring Cloud Stream 提供了多种灵活的机制来处理它们。需要注意的是，这些技术依赖于绑定器（binder）的实现、底层消息中间件的能力以及编程模型（稍后会详细讨论）。
每当消息处理器（函数）抛出异常时，该异常会传播回绑定器（binder），此时绑定器会使用 Spring Retry 库提供的 RetryTemplate 对同一条消息进行多次重试（默认重试 3 次）。如果重试失败，则由错误处理机制决定如何处理，可能会丢弃消息、重新排队消息以进行重新处理，或者将失败的消息发送到死信队列（DLQ）。
Rabbit 和 Kafka 都支持这些概念（尤其是 DLQ）。然而，其他绑定器可能不支持，因此请参考您所使用的绑定器的文档，以了解支持的错误处理选项的详细信息。
但请注意，reactive function 并不符合消息处理器的定义，因为它不处理单个消息，而是提供了一种将框架提供的流（即 Flux）与用户提供的流连接起来的方式。为什么这很重要？ 这是因为在本节后面你读到的关于重试模板、丢弃失败消息、重试、死信队列（DLQ）以及相关的配置属性的内容 仅 适用于消息处理器（即命令式函数）。
Reactive API 提供了非常丰富的操作符和机制库，以帮助您处理特定于各种复杂反应式用例的错误处理，这些用例远比简单的消息处理程序情况复杂得多。因此，请使用它们，例如您可以在 reactor.core.publisher.Flux 中找到的 public final Flux<T> retryWhen(Retry retrySpec);。
@Bean
public Function<Flux<String>, Flux<String>> uppercase() {
return flux -> flux
.retryWhen(Retry.backoff(3, Duration.ofMillis(1000)))
.map(v -> v.toUpperCase());
}
丢弃失败的消息
默认情况下，系统提供了错误处理器。第一个错误处理器会简单地记录错误信息。第二个错误处理器是绑定器特定的错误处理器，负责在特定消息系统（例如发送到死信队列）的上下文中处理错误信息。但由于没有提供额外的错误处理配置（在当前场景下），这个处理器不会执行任何操作。因此，本质上在记录错误信息后，消息将被丢弃。
虽然在某些情况下是可以接受的，但在大多数情况下，这是不可行的，我们需要一些恢复机制来避免消息丢失。
处理错误信息
在上一节中我们提到，默认情况下，导致错误的消息会被有效记录并丢弃。该框架还提供了一种机制，允许你提供自定义的错误处理程序（例如，发送通知或写入数据库等）。你可以通过添加专门设计用于接受 ErrorMessage 的 Consumer 来实现这一点，ErrorMessage 除了包含有关错误的所有信息（例如，堆栈跟踪等）外，还包含原始消息（触发错误的消息）。注意：自定义错误处理程序与框架提供的错误处理程序（例如，日志记录和绑定器错误处理程序——参见上一节）是互斥的，以确保它们不会相互干扰。
@Bean
public Consumer<ErrorMessage> myErrorHandler() {
return v -> {
// send SMS notification code
};
}
要将此类消费者识别为错误处理程序，您只需提供指向函数名称的 error-handler-definition 属性 - spring.cloud.stream.bindings.<binding-name>.error-handler-definition=myErrorHandler。
例如，对于绑定名称 uppercase-in-0，属性将如下所示：
spring.cloud.stream.bindings.uppercase-in-0.error-handler-definition=myErrorHandler
如果你使用了特殊的映射指令将绑定映射到一个更易读的名称——spring.cloud.stream.function.bindings.uppercase-in-0=upper，那么这个属性将如下所示：
spring.cloud.stream.bindings.upper.error-handler-definition=myErrorHandler.
如果你不小心将这样的处理程序声明为 Function，它仍然可以工作，只是它的输出不会被处理。然而，鉴于这样的处理程序仍然依赖于 Spring Cloud Function 提供的功能，如果你的处理程序有一些复杂性，并且你希望通过函数组合来解决（尽管这种情况不太可能），你也可以从函数组合中受益。
默认错误处理器
如果你想为所有函数 Bean 使用单一的错误处理器，你可以使用标准的 Spring Cloud Stream 机制来定义默认属性 spring.cloud.stream.default.error-handler-definition=myErrorHandler。
DLQ - 死信队列
或许最常见的机制是 DLQ，它允许将失败的消息发送到一个特殊的目的地：死信队列。
配置后，失败的消息将发送到此目的地，以便进行后续的重新处理、审核和对账。
考虑以下示例：
@SpringBootApplication
public class SimpleStreamApplication {
public static void main(String[] args) throws Exception {
SpringApplication.run(SimpleStreamApplication.class,
"--spring.cloud.function.definition=uppercase",
"--spring.cloud.stream.bindings.uppercase-in-0.destination=uppercase",
"--spring.cloud.stream.bindings.uppercase-in-0.group=myGroup",
"--spring.cloud.stream.rabbit.bindings.uppercase-in-0.consumer.auto-bind-dlq=true"
);
}
@Bean
public Function<Person, Person> uppercase() {
return personIn -> {
throw new RuntimeException("intentional");
});
};
}
}
提醒一下，在这个例子中，uppercase-in-0 属性的 uppercase-in-0 部分对应输入目标绑定的名称。consumer 部分表示这是一个消费者属性。
在使用 DLQ 时，必须至少提供 group 属性，以便正确命名 DLQ 目标。然而，group 通常与 destination 属性一起使用，如我们的示例所示。
除了设置一些标准属性外，我们还设置了 auto-bind-dlq，以指示绑定器为 uppercase-in-0 绑定创建并配置 DLQ 目标，该目标对应于 uppercase 目的地（参见相关属性）。这将导致创建一个名为 uppercase.myGroup.dlq 的额外 RabbitMQ 队列（有关 Kafka 特定的 DLQ 属性，请参阅 Kafka 文档）。
配置完成后，所有失败的消息都会被路由到此目的地，并保留原始消息以便进一步操作。
你可以看到错误信息包含了与原始错误相关的更多信息，如下所示：
. . . .
x-exception-stacktrace: org.springframework.messaging.MessageHandlingException: nested exception is
org.springframework.messaging.MessagingException: has an error, failedMessage=GenericMessage [payload=byte[15],
headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedRoutingKey=input.hello, amqp_deliveryTag=1,
deliveryAttempt=3, amqp_consumerQueue=input.hello, amqp_redelivered=false, id=a15231e6-3f80-677b-5ad7-d4b1e61e486e,
amqp_consumerTag=amq.ctag-skBFapilvtZhDsn0k3ZmQg, contentType=application/json, timestamp=1522327846136}]
at org.spring...integ...han...MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:107)
at. . . . .
Payload: blah
你也可以通过将 max-attempts 设置为 '1' 来立即将消息发送到 DLQ（无需重试）。例如，
--spring.cloud.stream.bindings.uppercase-in-0.consumer.max-attempts=1
重试模板
在本节中，我们将介绍与重试功能配置相关的配置属性。
RetryTemplate 是 Spring Retry 库的一部分。虽然本文档的范围不涵盖 RetryTemplate 的所有功能，但我们将提到以下与 RetryTemplate 特别相关的消费者属性：
maxAttempts
处理消息的尝试次数。
默认值：3。
backOffInitialInterval
重试时的初始退避间隔。
默认 1000 毫秒。
backOffMaxInterval
最大回退间隔。
默认 10000 毫秒。
backOffMultiplier
退避乘数。
默认 2.0。
defaultRetryable
监听器抛出的未在 retryableExceptions 中列出的异常是否可重试。
默认值：true。
retryableExceptions
一个映射，其中键为 Throwable 类名，值为布尔类型。指定哪些异常（及其子类）将会或不会被重试。另请参阅 defaultRetriable。示例：spring.cloud.stream.bindings.input.consumer.retryable-exceptions.java.lang.IllegalStateException=false。
默认值：空。
虽然前面的设置足以满足大多数自定义需求，但它们可能无法满足某些复杂需求，此时你可能希望提供自己的 RetryTemplate 实例。为此，请在你的应用程序配置中将其配置为一个 bean。应用程序提供的实例将覆盖框架提供的实例。此外，为了避免冲突，你必须将希望绑定器使用的 RetryTemplate 实例限定为 @StreamRetryTemplate。例如，
@StreamRetryTemplate
public RetryTemplate myRetryTemplate() {
return new RetryTemplate();
}
从上面的示例中可以看出，你不需要使用 @Bean 来注解它，因为 @StreamRetryTemplate 是一个合格的 @Bean。
如果你需要更精确地配置 RetryTemplate，可以在 ConsumerProperties 中通过名称指定 bean，以便为每个绑定关联特定的重试 bean。
spring.cloud.stream.bindings.<foo>.consumer.retry-template-name=<your-retry-template-bean-name>