6.3. 死信频道

概述

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

图 6.3. 死的频道模式

死信频道模式

在 Java DSL 中创建死信频道

以下示例演示了如何使用 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 示例

您可以在 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 “重新传送策略设置” 总结了可用于设置重新传送策略的方法。

表 6.1. 重新传送策略设置

方法签名默认描述

allowRedeliveryWhileStopping()

true

控制在安全关闭期间或路由停止期间是否尝试重新发送。在启动停止时,已经进行的交付不会中断。

BackOffMultiplier(倍数)

2

如果启用了 exponential backoff,让 m 设置为 backoff 数,并让 d 为初始延迟。然后重新发送尝试的顺序如下:

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

collisionAvoidancePercent(double collisionAvoidancePercent)

15

如果启用了冲突,请 冲突避免了百分比。冲突策略随后调整下一个延迟,最大为当前值的加号/减去 p%

deadLetterHandleNewException

true

Camel 2.15:指定是否在死信频道中处理消息时是否处理异常。如果为 true,则会处理异常并记录在 WARN 级别(因此保证死信频道完成)。如果为 false,则异常不会处理,因此死信频道失败,并传播新的异常。

delayPattern(String delayPattern)

Apache Camel 2.0:请参阅 “redeliver delay pattern”一节

disableRedelivery()

true

Apache Camel 2.0:禁用重新传送功能。要启用重新传送,将 maximumRedeliveries() 设置为正整数值。

handled(布尔值已处理)

true

Apache Camel 2.0:如果为 true,则当消息移动到 dead letter 频道时清除当前异常;如果为 false,则异常将传回给客户端。

initialRedeliveryDelay(long initialRedeliveryDelay)

1000

指定在尝试第一个重新发送前的延迟(以毫秒为单位)。

logNewException

true

指定是否在 WARN 级别登录,当在死信频道引发异常时。

logStackTrace(boolean logStackTrace)

false

Apache Camel 2.0:如果为 true,则 JVM 堆栈追踪包含在错误日志中。

maximumRedeliveries(int maximumRedeliveries)

0

Apache Camel 2.0:最大交付尝试次数.

maximumRedeliveryDelay(long maxDelay)

60000

Apache Camel 2.0:当使用指数级 backoff 策略时(请参阅 useExponentialBackOff())时,重建延迟可能会在不限制的情况下增加。此属性对重新发送延迟(以毫秒为单位)实施上限。

onRedelivery(处理器)

Apache Camel 2.0:配置在每次重新发送尝试之前调用的处理器。

redeliveryDelay(long int)

0

Apache Camel 2.0:指定重新传送尝试之间的延时(以毫秒为单位)。Apache Camel 2.16.0 : 默认重新传送延迟为一秒钟。

retriesExhaustedLogLevel(LoggingLevel logLevel)

LoggingLevel.ERROR

Apache Camel 2.0:指定日志记录级别,在其中记录发送失败(指定为 org.apache.camel.LoggingLevel constant)。

retryAttemptedLogLevel(LoggingLevel logLevel)

LoggingLevel.DEBUG

Apache Camel 2.0:指定日志记录级别,在其中重新发送尝试(指定为 org.apache.camel.LoggingLevel constant)。

useCollisionAvoidance()

false

启用冲突可避免,这会在 backoff 计时中添加一些随机性,以减少竞争概率。

useOriginalMessage()

false

Apache Camel 2.0:如果启用了此功能,则发送到 dead letter 频道的消息是 原始消息 交换的副本,因为它存在于路由的开头(在 from() 节点上)。

useExponentialBackOff()

false

启用 exponential backoff。

重新传送标头

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

表 6.2. 死 Letter Redelivery Headers

标头名称类型描述

CamelRedeliveryCounter

整数

Apache Camel 2.0:计数发送尝试失败的次数。Exchange.REDELIVERY_COUNTER 中也设置了这个值。

CamelRedelivered

布尔值

Apache Camel 2.0: True(如果进行了一个或多个重新发送尝试)。此值在 Exchange.REDELIVERED 中设定。

CamelRedeliveryMaxCounter

整数

Apache Camel 2.6:包含最大重新传送设置(也在 Exchange.REDELIVERY_MAX_COUNTER 交换属性中设置)。如果使用 retryWhile,或者配置了无限重新传送,则缺少此标头。

重新传送交换属性

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

表 6.3. 重新传送交换属性

Exchange Property Name类型描述

Exchange.FAILURE_ROUTE_ID

字符串

提供失败的路由 ID。这个属性的字面名称是 CamelFailureRouteId

使用原始消息

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

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

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

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

redeliver delay pattern

可作为 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.4 = 0 毫秒(作为第一个组以 5 开始)。
  • 尝试编号 5..9 = 1000 毫秒(第一个组)。
  • 尝试编号 10.19 = 5000 毫秒(第二个组)。
  • 尝试编号 20.. = 20000 毫秒(最后一个组)。

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

  • 尝试编号 1.4 = 1000 millis(第一个组)
  • 尝试编号 5.. = 5000 millis(最后一个组)

不要求下一个延迟应高于上述延迟,您可以使用您喜欢的任何延迟值。例如,延迟模式 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");

现在假定 foo an 中出现失败。在这种情况下,Exchange.TO_ENDPOINT 属性和 Exchange.FAILURE_ENDPOINT 属性仍包含该值。

onRedelivery 处理器

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

例如,以下死信频道被配置为在红色发生交换前调用 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));
注意

出于向后兼容的原因,allowRedeliveryWhileStopping 选项默认为 true。但是,在主动关闭过程中,重新传送始终会被隐藏,这与这个选项设置无关(例如,安全关闭超时)。

使用ExceptionOccurred Processor

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

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

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

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

onException 子句

您无需在路由构建器中使用 errorHandler() 拦截器,您可以定义一系列 onException() 子句,以定义不同的重新传送策略,以及为各种异常类型定义不同的死信频道。例如,若要为每个 NullPointerExceptionIOExceptionException 类型定义不同的行为,您可以使用 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 的消息标头中记录一些错误详情。

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

OnPrepareFailure

在将交换传递给死信队列之前,您可以使用 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"/>