Apache Camel - 具有死信通道功能的事务错误处理程序

Question

我正在开发一个消息路由器，我来处理意外情况并进行一些错误处理。

在我的代码中，我能够区分何时因错误消息而发生异常，这是我想将消息发送到死信端点的唯一一种情况。 在所有其他情况下，我会考虑由基础设施问题（例如，数据库/JMS 端点变得不可用）引起的异常，在这种情况下，我希望消息回滚到路由开始。

查看 Camel 文档，唯一支持死信的错误处理程序是 DeadLetterChannel，但问题是此错误处理程序未处理。

那么有没有一种方法可以实现我想要的，以及会有多简单或多难？我可以看到您可以配置骆驼上下文以使用自定义错误处理程序构建器，并且正在考虑在我自己的构建器中尝试在 TransactionErrorHandler 和 DeadLetterChannel 之间进行组合，但我不确定这是否是可行的方法。开箱即用的构建器似乎有相当多的复杂逻辑。

另一种选择是扩展TransactionErrorHandlerBuilder 并从DeadLetterChannelBuilder 引入验证端点并创建故障处理器但再次不确定的代码。如果这很容易，Camel 的人就会将它包含在框架中。我的用例必须是处理企业关键应用程序的任何人的用例。

提前感谢您的建议。任何提示将不胜感激。

更新 如上所述，我尝试扩展 TransactionErrorHandlerBuilder，但它没有工作，因为它创建的故障处理器从未使用过。

因为我真的需要这个功能，所以我尝试通过让我的代码在消息头中添加一个 **"dead.letter=true" 并将原始消息放回交换中，如下所示：

@Override
public void process(Exchange exchange) {
    Message incomingMessage = exchange.getIn();
    try {
        // Do some work here
    } catch (MyCustomException e) {    
        incomingMessage.setHeader("dead.letter", "true");
        exchange.setIn(incomingMessage);
    } catch (Exception e) {
        exchange.setException(e);
    }
}

然后在我的路由定义中添加：

from(ep).routeId(createRouteId(system))
        .autoStartup(false).transacted()
        .threads(route.getThreads())
        .filter(body().isNotNull())
        .process((Processor) routeBean)
        .choice()
        .when(header("dead.letter").isNotNull())
        .to(mq1:ERROR.QUEUE);

我希望这会解决我的问题。但是我的 ERROR.QUEUE 没有得到任何东西（日志中也没有任何东西）并且事务提交并且我的消息丢失了。

请帮忙，因为我的想法不多了。

Answer 1

我遇到了同样的问题。我不知道为什么不能将 TransactionErrorHandlerBuilder 配置为将消息发送到死信通道。 我被它的方法 setDeadLetterUri() 弄糊涂了，它不像我预期的那样工作。经过长时间的研究，我找到了相当简单的解决方案。

它在事务中处理消息，如果抛出一些异常，事务将被回滚，原始消息将被发送到死信通道。 请注意，您应该明确允许使用原始消息。

@Override
public void configure() {
    
    getContext().setAllowUseOriginalMessage(true); 

    from("jms:queue:{{foo.bar}}")
            .doTry()
                .to("direct:processMsgInTransaction")
            .doCatch(Exception.class)
                .process(exchange -> exchange.setMessage(exchange.getUnitOfWork().getOriginalInMessage()))
                .to("jms:queue:{{foo.bar.dead.letter}}")
            .endDoTry();

    from("direct:processMsgInTransaction")
            .errorHandler(new TransactionErrorHandlerBuilder() 
                .onExceptionOccurred(exchange -> log.error("Something went wrong")))
            //.transacted() //uncomment for using default TransactionErrorHandler
            .process(someProcessor)
            .process(someOtherProcessor);
            
}