错误处理
在本节中,我们将解释框架提供的错误处理机制背后的一般思想。 我们将使用 Rabbit 活页夹作为示例,因为各个活页夹定义了不同的集合 特定于底层代理功能(例如 Kafka 绑定器)的某些受支持机制的属性。
错误时有发生,Spring Cloud Stream 提供了几种灵活的机制来处理它们。请注意,这些技术取决于 binder 实现和 底层消息传递中间件的功能以及编程模型(稍后会详细介绍)。
每当消息处理程序(函数)抛出异常时,它都会传播回绑定器,此时绑定器将多次尝试重试
相同的消息(默认为 3)使用RetryTemplate
由 Spring Retry 库提供。
如果重试不成功,则取决于错误处理机制,该机制可能会删除消息、将消息重新排队以重新处理或将失败的消息发送到 DLQ。
Rabbit 和 Kafka 都支持这些概念(尤其是 DLQ)。但是,其他活页夹可能没有,因此请参阅各个活页夹的文档,了解有关支持的详细信息 错误处理选项。
但是请记住,响应式函数不符合 Message 处理程序的条件,因为它不处理单个消息和 相反,提供了一种将框架提供的流(即 Flux)与用户提供的流连接起来的方法。为什么这很重要?这是因为您在本节后面阅读的有关重试模板、删除失败消息、重试、 帮助完成所有这些作的 DLQ 和配置属性仅适用于消息处理程序(即命令式函数)。
Reactive API 提供了一个非常丰富的自己的运算符和机制库,以帮助您处理特定于
各种响应式用例,它们比简单的消息处理程序情况复杂得多,所以使用它们,例如
如public final Flux<T> retryWhen(Retry retrySpec);
你可以在reactor.core.publisher.Flux
.
@Bean
public Function<Flux<String>, Flux<String>> uppercase() {
return flux -> flux
.retryWhen(Retry.backoff(3, Duration.ofMillis(1000)))
.map(v -> v.toUpperCase());
}
丢弃失败的邮件
默认情况下,系统提供错误处理程序。第一个错误处理程序将简单地记录错误消息。第二个错误处理程序是特定于绑定程序的错误处理程序 它负责在特定消息传递系统的上下文中处理错误消息(例如,发送到 DLQ)。但是,由于没有提供其他错误处理配置(在当前情况下),因此此处理程序将不会执行任何作。因此,基本上在记录后,消息将被删除。
虽然在某些情况下是可以接受的,但在大多数情况下,它不是,我们需要一些恢复机制来避免消息丢失。
处理错误消息
在上一节中,我们提到,默认情况下,导致错误的消息会被有效地记录和删除。框架还为你公开了机制
提供自定义错误处理程序(即发送通知或写入数据库等)。您可以通过添加Consumer
专门设计用于接受ErrorMessage
除了有关错误的所有信息(例如,堆栈跟踪等)之外,还包含原始消息(触发错误的消息)。
自定义错误处理程序与框架提供的错误处理程序(即日志记录错误处理程序和特定于绑定程序的错误处理程序 - 请参阅上一节)互斥,以确保它们不会干扰。提供自定义错误处理程序时,即使 DLQ 配置为这样做,向 DLQ 发送失败的消息也无法工作。 |
@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
segment 对应于输入目标绑定的名称。
这consumer
segment 表示它是消费者属性。
使用 DLQ 时,至少group 必须提供属性才能正确命名 DLQ 目标。然而group 经常一起使用
跟destination 属性,如我们的示例所示。 |
除了一些标准属性之外,我们还设置了auto-bind-dlq
指示活页夹创建和配置 DLQ 目标uppercase-in-0
binding 对应于uppercase
destination(请参阅相应的属性),这会导致一个名为uppercase.myGroup.dlq
(有关 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”。例如
--spring.cloud.stream.bindings.uppercase-in-0.consumer.max-attempts=1
重试模板
在本节中,我们将介绍与重试功能配置相关的配置属性。
这RetryTemplate
是 Spring Retry 库的一部分。
虽然它超出了本文档的范围,但涵盖RetryTemplate
我们
会提到以下具体相关的消费属性
这RetryTemplate
:
- 最大尝试次数
-
尝试处理邮件的次数。
默认值:3。
- backOff初始间隔
-
重试时的回退初始间隔。
默认为 1000 毫秒。
- backOffMax间隔
-
最大回退间隔。
默认为 10000 毫秒。
- backOff乘数
-
退避乘数。
默认 2.0。
- default可重试
-
监听器抛出的异常是否未在
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。
spring.cloud.stream.bindings.<foo>.consumer.retry-template-name=<your-retry-template-bean-name>