错误处理器
错误处理器
Camel 支持可插拔的 ErrorHandler 策略,用于处理事件驱动消费者在处理过程中出现的错误。
异常子句
将错误处理器与异常子句结合使用是一种非常强大的组合。我们建议最终用户在错误处理策略中使用这种组合。请参阅示例和异常子句。
使用 try … catch … finally
与错误处理相关的是可以直接在路由中使用的 Try Catch Finally DSL。它基本上是 Java 语言中常规 try catch finally 的模拟,但功能更强大。
Camel 当前开箱即提供的实现如下:
非事务性
- DefaultErrorHandler 是 Camel 中的默认错误处理器。该错误处理器不支持死信队列,它会将异常传播回调用方,就像根本没有错误处理器一样。它的功能集较为有限。
- Dead Letter Channel(死信通道),支持在将消息交换发送到死信端点之前尝试重新投递若干次
- NoErrorHandler,用于不进行任何错误处理
事务性
- TransactionErrorHandler 是 Camel 中用于事务路由的默认错误处理器。请参阅事务性客户端 EIP 模式。
这些错误处理器可以在 DSL 中应用于整组规则或特定的路由规则,我们将在接下来的示例中进行演示。在单个 RouteBuilder 中,错误处理规则会被其内的每条路由规则继承。
所提供的错误处理器简要总结
DefaultErrorHandler
DefaultErrorHandler 是 Camel 中的默认错误处理器。与 Dead Letter Channel 不同,它没有任何死信队列,并且默认情况下不会处理异常。
Dead Letter Channel
Dead Letter Channel 最多会以 1 秒的延迟重新投递 6 次,如果交换失败,将以 ERROR 级别记录日志。
你可以配置默认使用的死信端点:或在 Spring XML DSL 中配置:
Spring XML
<camel:errorHandler id="deadLetterErrorHandler" type="DeadLetterChannel" deadLetterUri="log:dead">
<camel:camelContext errorHandlerRef="deadLetterErrorHandler">
...
</camel:camelContext>无错误处理器
无错误处理器用于禁用错误处理。
- Java
- Spring XML
errorHandler(noErrorHandler());<camel:errorHandler id="noErrorHandler" type="NoErrorHandler"/>
<camel:camelContext errorHandlerRef="noErrorHandler">
...
</camel:camelContext>TransactionErrorHandler
TransactionErrorHandler 是 Camel 中事务路由的默认错误处理器。
如果你使用 transacted DSL 将路由标记为事务路由,Camel 会自动使用 TransactionErrorHandler。它会尝试查找全局/按路由配置的错误处理器,如果该处理器是 TransactionErrorHandlerBuilder 实例则直接使用;如果不是,Camel 会自动创建一个临时的 TransactionErrorHandler 并覆盖默认错误处理器。这就是「约定优于配置」的体现。 |
|---|
各种错误处理器支持的特性
以下是各错误处理器所支持特性的详细说明:
| 特性 | 支持该特性的错误处理器 |
|---|---|
| 所有作用域 | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| onException | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| onWhen | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| continued | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| handled | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| 自定义 ExceptionPolicy | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| useOriginalBody | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| retryWhile | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| onRedelivery | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| RedeliveryPolicy | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| asyncDelayedRedelivery | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| redeliverWhileStopping | DefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel |
| 死信队列 | Dead Letter Channel |
| onPrepareFailure | DefaultErrorHandler、Dead Letter Channel |
有关上述特性的更多文档,请参阅 异常子句 文档。
作用域
错误处理器的作用域可以是:
- CamelContext - 在 XML 中是全局的,而在 Java DSL 中则是仅限于同一个
RouteBuilder内的全局 - Route - 每条路由各自独立
下面的示例展示了如何为 RouteBuilder 注册一个全局错误处理器:
纯 Java:带有全局错误处理器的 RouteBuilder
RouteBuilder builder = new RouteBuilder() {
public void configure() {
errorHandler(deadLetterChannel("seda:error"));
// here is our regular route
from("seda:a").to("seda:b");
}
};以下示例演示了如何注册特定于路由的错误处理器:
仅 Java:带有特定于路由错误处理器的 RouteBuilder
RouteBuilder builder = new RouteBuilder() {
public void configure() {
// this route is using a nested error handler
from("seda:a")
// here we configure the error handler
.errorHandler(deadLetterChannel("seda:error"))
// and we continue with the routing here
.to("seda:b");
// this route will use the default error handler
from("seda:b").to("seda:c");
}
};基于 XML 和 YAML 的配置
Java DSL 与 Spring DSL 的区别 错误处理器在 Java DSL 和 XML/YAML DSL 中的配置方式略有不同。传统的 Spring XML 更多依赖标准的 Spring <beans> XML 配置,而 Java DSL 使用流式构建器。
错误处理器可以作为传统的 Spring XML Bean 配置,并可限定作用范围:
- 全局(
<camelContext>标签) - 按路由(
<route>标签) - 或按策略(
<policy>/<transacted>标签)
错误处理器通过 errorHandlerRef 属性进行配置。
错误处理器层级
错误处理器是继承的,因此如果你只设置了全局错误处理器,它将处处生效。但你可以在某条路由中覆盖它,改用其他的错误处理器。
在下面的示例中,我们在路由上配置了一个死信通道,该通道最多重新投递 3 次,并在重试前加入短暂的延迟。
- XML
- 传统 Spring XML
- YAML
<route>
<errorHandler>
<deadLetterChannel deadLetterUri="mock:dead">
<redeliveryPolicy maximumRedeliveries="3" redeliveryDelay="250"/>
</deadLetterChannel>
</errorHandler>
<from uri="direct:in"/>
<process ref="myFailureProcessor"/>
<to uri="mock:result"/>
</route>首先,我们使用 route 标签上的 errorHandlerRef 属性来配置对 myDeadLetterErrorHandler 的引用。
然后,我们配置 myDeadLetterErrorHandler,它就是我们的死信通道。此配置是使用 bean 元素的旧式 Spring XML。最后,我们还有一个用于重新投递策略的 Spring bean,在其中可以配置重新投递次数、延迟等选项。
<!-- here we configure our DeadLetterChannel -->
<bean id="myDeadLetterErrorHandler" class="org.apache.camel.builder.DeadLetterChannelBuilder">
<!-- exchanges is routed to mock:dead in cased redelivery failed -->
<property name="deadLetterUri" value="mock:dead"/>
<!-- reference the redelivery policy to use -->
<property name="redeliveryPolicy" ref="myRedeliveryPolicyConfig"/>
</bean>
<!-- here we set the redelivery settings -->
<bean id="myRedeliveryPolicyConfig" class="org.apache.camel.processor.errorhandler.RedeliveryPolicy">
<!-- try redelivery at most 3 times, after that the exchange is dead and its routed to the mock:dead endpoint -->
<property name="maximumRedeliveries" value="3"/>
<!-- delay 250ms before redelivery -->
<property name="redeliveryDelay" value="250"/>
</bean>
<camelContext xmlns="http://camel.apache.org/schema/spring">
<!-- set the errorHandlerRef to our DeadLetterChannel, this applies for this route only -->
<route errorHandlerRef="myDeadLetterErrorHandler">
<from uri="direct:in"/>
<process ref="myFailureProcessor"/>
<to uri="mock:result"/>
</route>
</camelContext>- route:
errorHandler:
deadLetterChannel:
deadLetterUri: mock:dead
redeliveryPolicy:
maximumRedeliveries: 3
redeliveryDelay: 250
from:
uri: direct:in
steps:
- process:
ref: myFailureProcessor
- to:
uri: mock:result使用事务错误处理器
事务错误处理器基于 Spring 事务。这需要使用 camel-spring 或 camel-jta 组件。
参见事务客户端,其中提供了许多关于如何使用该错误处理器、其事务行为以及相关配置的示例。
如何无限重试失败的消息?
如果你希望让出错的消息保留在原系统中(例如消息中间件),那么在该出错消息之后到达队列的后续消息也会被阻塞。
例如使用 ActiveMQ 时,Camel 会重试消费某条消息最多 6 次,之后该消息会被移动到 Broker 的默认死信队列(这是 ActiveMQ Broker 特有的行为)。
如果你将默认错误处理器或死信通道配置为使用 maximumRedeliveries = -1,那么 Camel 将会无限重试。
| 注意,让 Camel 重试处理消息(例如 Camel 将消息路由到外部系统(生产者)时失败)与从支持重新投递的消息系统(如 ActiveMQ)消费消息是有区别的。后者允许回滚整条消息,从而让 Camel 再次尝试处理同一条消息。 |
|---|
如果像 ActiveMQ 这样的外部系统正在重新投递某条消息,Camel 会为该消息添加用于标记这一情况的头部信息。
CamelRedeliveryCounter 包含该消息已被重新投递的次数。CamelRedelivered 是一个布尔值,表示消息是被重新投递的还是首次被处理。
另请参见事务客户端。
如何从某个点或整条路由回退重试处理消息
默认情况下,Camel 会从故障发生的点开始执行任何重新投递(重试)操作。因此,如果你想从更早的点开始重试,就需要拆分你的路由。
在下面的示例中,我们有 2 条路由(direct:start 和 direct:sub)。如果 direct:sub 路由中的任何位置发生故障,整条路由都会被重试。这是因为我们指示 direct:sub 路由不使用任何错误处理器(例如无错误处理器)。然后,我们通过从第一条路由调用子路由,使用 Direct 组件将这些路由连接起来。
- Java
- Spring XML
- XML
- YAML
// in case of io exception then try to redeliver up till 2 times
// (do not use any delay due faster unit testing)
onException(IOException.class)
.maximumRedeliveries(2).redeliveryDelay(0);
from("direct:start")
.to("mock:a")
// call sub route (using direct)
.to("direct:sub")
.to("mock:c");
from("direct:sub")
// disable error handler, so the entire route can be retried in case of redelivery
.errorHandler(noErrorHandler())
.to("mock:b")
.bean("myProcessor");<!-- this is the processor that will fail the first 2 attempts -->
<bean id="myProcessor" class="org.apache.camel.processor.RedeliverToSubRouteTest.MyProcessor"/>
<camelContext xmlns="http://camel.apache.org/schema/spring">
<!-- setup no error handler with an id, we refer to from the 2nd route -->
<errorHandler id="noErrorHandler" type="NoErrorHandler"/>
<!-- configure on exception to redelivery at most 2 times when an IOException was thrown
do not use redelivery delay to run unit test faster -->
<onException>
<exception>java.io.IOException</exception>
<redeliveryPolicy maximumRedeliveries="2" redeliveryDelay="0"/>
</onException>
<!-- 1st route, no need to setup error handler, as it will use the default error handler -->
<route>
<from uri="direct:start"/>
<to uri="mock:a"/>
<to uri="direct:sub"/>
<to uri="mock:c"/>
</route>
<!-- disable error handler on this route, so the entire route can be redelivered
when called from the 1st route -->
<route errorHandlerRef="noErrorHandler">
<from uri="direct:sub"/>
<to uri="mock:b"/>
<process ref="myProcessor"/>
</route>
</camelContext><route>
<from uri="direct:start"/>
<onException>
<exception>java.io.IOException</exception>
<redeliveryPolicy maximumRedeliveries="2" redeliveryDelay="0"/>
</onException>
<to uri="mock:a"/>
<to uri="direct:sub"/>
<to uri="mock:c"/>
</route>
<route>
<errorHandler>
<noErrorHandler/>
</errorHandler>
<from uri="direct:sub"/>
<to uri="mock:b"/>
<process ref="myProcessor"/>
</route>- onException:
# in case of io exception then try to redeliver up till 2 times
# (do not use any delay due faster unit testing)
exception:
- java.io.IOException
redeliveryPolicy:
maximumRedeliveries: 2
redeliveryDelay: 0
- route:
from:
uri: direct:start
steps:
- to:
uri: mock:a
- to:
uri: direct:sub
- to:
uri: mock:c
- route:
errorHandler:
noErrorHandler: {}
from:
uri: direct:sub
steps:
- to:
uri: mock:b
- process:
ref: myProcessor上面的代码基于一个单元测试,可以看到下面的处理器被配置为在前两次尝试时失败。这意味着整个 direct:sub 路由都会被重新投递,也就是说 mock:b 端点会再次收到传入的消息。
public static class MyProcessor implements Processor {
private int counter;
@Override
public void process(Exchange exchange) throws Exception {
// use a processor to simulate error in the first 2 calls
if (counter++ < 2) {
throw new IOException("Forced");
}
exchange.getIn().setBody("Bye World");
}
}评论
登录后参与评论
KnowForge