使用特定于路由的 DLQ 配置 Java/Camel/AMQ答案

【问题标题】：Configuring Java/Camel/AMQ with route-specific DLQs使用特定于路由的 DLQ 配置 Java/Camel/AMQ
【发布时间】：2019-03-02 15:54:21
【问题描述】：

此处为 Java 8/Camel 2.19.x/AMQ 5.15.x。

我有一个 Java 应用程序，它使用 Camel 来使用 AMQ 队列中的消息、处理这些消息并用它们做一些事情。有时，路由的输出会将处理结果备份到另一个队列中以进行进一步的下游处理，但并非总是/必要。典型的 Java/Camel/AMQ 设置。

我的每条路线（我使用 Camel XML DSL）都有一个配置的 <onException> 处理程序，通常如下所示：

<onException useOriginalMessage="true">
  <exception>java.lang.Exception</exception>

  <redeliveryPolicy logStackTrace="true"/>

  <handled>
    <constant>true</constant>
  </handled>

  <log message="${exception.stacktrace}" loggingLevel="ERROR"/>

  <rollback markRollbackOnly="true"/>

</onException>

非常简单：记录异常并回滚。

作为 <onException> 处理程序的一部分，我想做的是放置 原始消息（失败并导致抛出异常，不是例外！）在特定于路由的 DLQ 上（“DLQ”是指可以发送失败消息用于审计/报告/播放目的的队列）

意思是，如果我的应用有 30 个路由，每个路由从 30 个不同的 AMQ 队列消费，我将有 30 个不同的“DLQ”，它们各自的 <onException> 处理程序会将失败的消息发送到。

理想情况下，我希望此配置位于 AMQ 端（同样可能在 activem.xml 或类似内部），这样如果 DLQ 目标需要更改，我就不需要进行代码更改或重新部署。但是，如果只能从 Camel 路线/配置内部进行，那也没关系。

我假设我可以修改每个路由以包含其自己为原始消息定制的目标 DLQ：

<onException useOriginalMessage="true">
  <exception>java.lang.Exception</exception>

  <redeliveryPolicy logStackTrace="true"/>

  <handled>
    <constant>true</constant>
  </handled>

  <log message="${exception.stacktrace}" loggingLevel="ERROR"/>

  <rollback markRollbackOnly="true"/>

  <to uri="activemq:fizzbuzz.dlq"/>

</onException>

但我希望有比这更优雅的东西......

有什么想法可以做到这一点吗？

【问题讨论】：

标签： java apache-camel activemq dead-letter

【解决方案1】：

也许这种方式对你会更好：

             DeadLetterChannelBuilder errorHandlerBuilder = deadLetterChannel("jms:dummy");
    errorHandlerBuilder.onPrepareFailure(exchange -> {
        exchange.getIn().setHeader("CamelJmsDestinationName",exchange.getIn().getHeader("JMSDestination",String.class).concat(".DLQ"));
    });

    from("jms:input1")
            .to("seda:process");

    from("jms:input2")
            .to("seda:process");

    from("jms:input3")
            .to("seda:process");

    from("seda:process").errorHandler(errorHandlerBuilder)
            .process(exchange -> {
                throw new RuntimeException();
            });

您可以在运行时计算 DLQ 队列名称。 DeadLetterChannelBuilder 也可以像你的 onException 一样配置。

【讨论】：

Thans @c0old (+1) - 这是一个有趣的解决方案...虽然很快，JmsDestination 标头和 CamelJmsDestinationName 标头之间的区别是什么？再次感谢！！！
根据 JMS 文档“收到消息时，其 JMSDestination 值必须等于发送时分配的值。”。 CamelJmsDestinationName 在发送期间用于动态计算的队列名称。 JmsProducer 将使用此标头中的值进行发送，而不是在端点中指定。
测试后队列看起来像 this 。 “queue://input1”是 JMSDestination 标头的值（根据“input1”消费者）

【解决方案2】：

如果您想在 ActiveMQ 本身中进行配置，您还可以添加特定的 policyEntry 并在 deadLetterStrategy 中使用通配符，如下例所示：

为所有INBOUND队列创建一个专用的DLQ：

<policyEntry queue="*.INBOUND.>">
    <deadLetterStrategy>
        <individualDeadLetterStrategy processExpired="false" queuePrefix="" queueSuffix=".DLQ" useQueueForQueueMessages="true"/>
    </deadLetterStrategy>
</policyEntry>

在这种情况下，我在所有入站队列中都发现了消费者的错误：

XXX.INBOUND.AAA        -> XXX.INBOUND.AAA.DLQ
YYY.INBOUND            -> YYY.INBOUND.DLQ
ZZZ.INBOUND.BBB.CCC    -> ZZZ.INBOUND.BBB.CCC.DLQ

但是

NNN.MMM.INBOUND        -> ActiveMQ.DLQ

因为模式* 与NNN.MMM 中的点. 字符不匹配。

您可以根据您的用例调整您的模式。

这对于管理未正确捕获异常的消费者非常有用，这是在 ActiveMQ 级别完成的。

【讨论】：