6.3. 死信频道

死信频道 模式（如 图 6.3 “死的频道模式” 所示）描述了当消息系统无法向预期收件人发送邮件时要执行的操作。这包括重试交付等功能，如果交付最终失败，则向死信频道发送消息，以存档未发送的消息。

图 6.3. 死的频道模式

以下示例演示了如何使用 Java DSL 创建死信频道：

errorHandler(deadLetterChannel("seda:errors"));
from("seda:a").to("seda:b");

其中 errorHandler（） 方法是一个 Java DSL 拦截器，这意味着当前路由构建器中定义的所有路由都受到此设置的影响。deadLetterChannel（） 方法是一个 Java DSL 命令，它使用指定的目标端点 seda:errors 创建新的死信频道。

errorHandler（） 拦截器提供了一种 catch- all 机制，用于处理所有错误类型。如果要应用更精细的方法来例外处理，您可以使用 onException 子句（请参阅 “onException 子句”一节）。

您可以在 XML DSL 中定义死信频道，如下所示：

 <route errorHandlerRef="myDeadLetterErrorHandler">
    ...
 </route>

 <bean id="myDeadLetterErrorHandler" class="org.apache.camel.builder.DeadLetterChannelBuilder">
     <property name="deadLetterUri" value="jms:queue:dead"/>
     <property name="redeliveryPolicy" ref="myRedeliveryPolicyConfig"/>
 </bean>

 <bean id="myRedeliveryPolicyConfig" class="org.apache.camel.processor.RedeliveryPolicy">
     <property name="maximumRedeliveries" value="3"/>
     <property name="redeliveryDelay" value="5000"/>
 </bean>

通常，如果发送尝试失败，您不会直接向 dead letter 频道发送邮件。相反，您会重新尝试交付到一些最大限制，且在所有重新发送尝试失败时，您会将消息发送到 dead letter 频道。要自定义消息重新传送，您可以将死信频道配置为具有 重新传送策略。例如，要指定最多两个重新传送尝试，并在交付尝试之间应用 exponential backoff 算法到时间延迟，您可以配置死信频道，如下所示：

errorHandler(deadLetterChannel("seda:errors").maximumRedeliveries(2).useExponentialBackOff());
from("seda:a").to("seda:b");

您可以通过调用链中的相关方法在 dead letter 频道上设置重新传送选项（链中每个方法都将引用当前 RedeliveryPolicy 对象）。表 6.1 “重新传送策略设置” 总结了可用于设置重新传送策略的方法。

d, m*d, m*m*d, m*m*m*d, ...

如果 Apache Camel 尝试重新发送一条信息，它会在 In 消息中自动设置 表 6.2 “死 Letter Redelivery Headers” 中描述的标头。

如果 Apache Camel 尝试恢复信息，它会自动设置 表 6.3 “重新传送交换属性” 中描述的交换属性。

自 Apache Camel 2.0 起，由于交换对象通过路由，因此交换对象会被修改，所以当引发异常时，这个交换不一定要存储在死信频道中的副本。在许多情况下，最好在路由开始时记录到达路由的消息，然后再记录路由受到任何转换的转换。例如，请考虑以下路由：

from("jms:queue:order:input")
       .to("bean:validateOrder");
       .to("bean:transformOrder")
       .to("bean:handleOrder");

前面的路由会侦听传入的 JMS 消息，然后使用 Beans 序列处理消息： validateOrder、transformOrder 和 handleOrder。但是，当发生错误时，我们不知道该消息所处的状态。转换 操作程序或之后是否发生错误？我们可以通过启用 useOriginalMessage 选项来确保来自 jms:queue:order:input 的原始消息记录到 dead letter 频道：

// will use original body
errorHandler(deadLetterChannel("jms:queue:dead")
       .useOriginalMessage().maximumRedeliveries(5).redeliveryDelay(5000);

可作为 Apache Camel 2.0 使用 delayPattern 选项为重新传送计数的特定范围指定延迟。delay 模式具有以下语法： limit1:delay1;limit2:delay2;limit3:delay3;…​ ，其中每个 delayN 应用到红色的 delayN PERC redeliveryCount < limitN+1

例如，假设模式为 5:1000;10:5000;20:20000，它定义了三个组，并导致以下重新发送延迟：

您可以启动具有限制 1 的组来定义启动延迟。例如，1:1000;5:5000 会产生以下重新传送延迟：

不要求下一个延迟应高于上述延迟，您可以使用您喜欢的任何延迟值。例如，延迟模式 1:5000;3:1000，以 5 秒延迟开头，然后将延迟减少到 1 秒。

当 Apache Camel 路由消息时，它会更新包含 Exchange 发送到 的最后一个 端点的 Exchange 属性。因此，您可以使用以下代码获取当前交换最新目的地的 URI：

// Java
String lastEndpointUri = exchange.getProperty(Exchange.TO_ENDPOINT, String.class);

这里的 Exchange.TO_ENDPOINT 是一个等于 CamelToEndpoint 的字符串。当 Camel 发送消息 到任何 端点时，都会更新此属性。

如果在路由过程中发生错误，并且交换被移到死信队列中，Apache Camel 将额外设置名为 CamelFailureEndpoint 的属性，该属性标识在发生错误之前交换的最后一个目的地。因此，您可以使用以下代码从死信队列中访问失败端点：

// Java
String failedEndpointUri = exchange.getProperty(Exchange.FAILURE_ENDPOINT, String.class);

这里的 Exchange.FAILURE_ENDPOINT 是一个字符串常量，相当于 CamelFailureEndpoint。

        from("activemq:queue:foo")
        .to("http://someserver/somepath")
        .beanRef("foo");

当死信频道执行红色时，可以在每次重新发送尝试 前 配置只执行的处理器。这可用于您需要在消息被红色前修改消息的情况。

例如，以下死信频道被配置为在红色发生交换前调用 MyRedeliverProcessor ：

// we configure our Dead Letter Channel to invoke
// MyRedeliveryProcessor before a redelivery is
// attempted. This allows us to alter the message before
errorHandler(deadLetterChannel("mock:error").maximumRedeliveries(5)
        .onRedelivery(new MyRedeliverProcessor())
        // setting delay to zero is just to make unit teting faster
        .redeliveryDelay(0L));

MyRedeliveryProcessor 进程在其中实现，如下所示：

// This is our processor that is executed before every redelivery attempt
// here we can do what we want in the java code, such as altering the message
public class MyRedeliverProcessor implements Processor {

    public void process(Exchange exchange) throws Exception {
        // the message is being redelivered so we can alter it

        // we just append the redelivery counter to the body
        // you can of course do all kind of stuff instead
        String body = exchange.getIn().getBody(String.class);
        int count = exchange.getIn().getHeader(Exchange.REDELIVERY_COUNTER, Integer.class);

        exchange.getIn().setBody(body + count);

        // the maximum redelivery was set to 5
        int max = exchange.getIn().getHeader(Exchange.REDELIVERY_MAX_COUNTER, Integer.class);
        assertEquals(5, max);
    }
}

如果您停止路由或启动安全关闭，则错误处理程序的默认行为将继续尝试重新发送。因为这通常不是所需的行为，所以可以选择通过在关闭或停止期间禁用重新传送，方法是将 allowRedeliveryWhileStopping 选项设置为 false，如下例所示：

errorHandler(deadLetterChannel("jms:queue:dead")
    .allowRedeliveryWhileStopping(false)
    .maximumRedeliveries(20)
    .redeliveryDelay(1000)
    .retryAttemptedLogLevel(LoggingLevel.INFO));

死 Letter 通道支持 onExceptionOccurred 处理器，允许在发生异常后对消息进行自定义处理。您还可以将其用于自定义日志记录。从 onExceptionOccurred 处理器中引发的新异常都会记录为 WARN 并忽略，而不是覆盖现有异常。

onRedelivery 处理器和在ExceptionOccurred processor 之间的区别在于，您可以在重新发送尝试前完全处理前一个。但是，在发生异常后不会立即发生。例如，如果您将错误处理程序配置为在重新传送尝试之间执行 5 秒的延迟，则稍后在出现异常后调用重新发送处理器。

以下示例说明了如何在出现异常时执行自定义日志记录。您需要配置 onExceptionOccurred 以使用自定义处理器。

errorHandler(defaultErrorHandler().maximumRedeliveries(3).redeliveryDelay(5000).onExceptionOccurred(myProcessor));

您无需在路由构建器中使用 errorHandler（） 拦截器，您可以定义一系列 onException（） 子句，以定义不同的重新传送策略，以及为各种异常类型定义不同的死信频道。例如，若要为每个 NullPointerException、IOException 和 Exception 类型定义不同的行为，您可以使用 Java DSL 在路由构建器中定义以下规则：

onException(NullPointerException.class)
    .maximumRedeliveries(1)
    .setHeader("messageInfo", "Oh dear! An NPE.")
    .to("mock:npe_error");

onException(IOException.class)
    .initialRedeliveryDelay(5000L)
    .maximumRedeliveries(3)
    .backOffMultiplier(1.0)
    .useExponentialBackOff()
    .setHeader("messageInfo", "Oh dear! Some kind of I/O exception.")
    .to("mock:io_error");

onException(Exception.class)
    .initialRedeliveryDelay(1000L)
    .maximumRedeliveries(2)
    .setHeader("messageInfo", "Oh dear! An exception.")
    .to("mock:error");

from("seda:a").to("seda:b");

如果通过串联重新传送策略方法来指定重新传送选项（如 表 6.1 “重新传送策略设置”中列出的），再使用 to（） DSL 命令指定 dead letter 频道的端点。您也可以在 Exception（） 子句中调用其他 Java DSL 命令。例如，前面的示例调用 setHeader（） 在名为 messageInfo 的消息标头中记录一些错误详情。

在本例中，NullPointerException 和 IOException 异常类型是特别配置的。所有其他例外类型由通用 Exception 异常拦截器进行处理。默认情况下，Apache Camel 应用最符合所引发异常异常的异常拦截器。如果无法找到完全匹配，它将尝试匹配最接近的基本类型，以此类推。最后，如果没有其他拦截器匹配，则 Exception 类型的拦截器与所有剩余的例外匹配。

在将交换传递给死信队列之前，您可以使用 onPrepare 选项来允许一个自定义处理器准备交换。它可让您添加有关交换的信息，如交换失败的原因。例如，以下处理器添加带有异常消息的标头。

public class MyPrepareProcessor implements Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        Exception cause = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class);
        exchange.getIn().setHeader("FailedBecause", cause.getMessage());
    }
}

您可以将错误处理程序配置为使用处理器，如下所示：

errorHandler(deadLetterChannel("jms:dead").onPrepareFailure(new MyPrepareProcessor()));

但是，Prepare 选项也可以使用默认错误处理程序来使用。

<bean id="myPrepare"
class="org.apache.camel.processor.DeadLetterChannelOnPrepareTest.MyPrepareProcessor"/>

<errorHandler id="dlc" type="DeadLetterChannel" deadLetterUri="jms:dead" onPrepareFailureRef="myPrepare"/>