架构

错误处理器

师成师成· 更新于 2026-09-28· 阅读 30 分钟· 0 次阅读

登录后可跨设备保存划线和私人笔记登录

错误处理器

Camel 支持可插拔的 ErrorHandler 策略,用于处理事件驱动消费者在处理过程中出现的错误。

另一种方式是使用异常子句直接在 DSL 中指定错误处理。

异常子句

将错误处理器与异常子句结合使用是一种非常强大的组合。我们建议最终用户在错误处理策略中使用这种组合。请参阅示例和异常子句。

使用 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
onExceptionDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
onWhenDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
continuedDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
handledDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
自定义 ExceptionPolicyDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
useOriginalBodyDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
retryWhileDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
onRedeliveryDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
RedeliveryPolicyDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
asyncDelayedRedeliveryDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
redeliverWhileStoppingDefaultErrorHandler、TransactionErrorHandler、Dead Letter Channel
死信队列Dead Letter Channel
onPrepareFailureDefaultErrorHandler、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");
    }
}

评论

登录后参与评论

正在加载评论…