骆驼消息重新传递未按预期运行

Question

我在 Camel 中有一条路线，我想在发生异常时重试，但我想设置一个属性，以便该路线第二次可以做一些稍微不同的事情，以尝试阻止在重试时再次发生错误。这是一条说明我目前正在尝试的想法的路线。

from("direct:onExceptionTest")
    .onException(Exception.class)
        .maximumRedeliveries(1)
        .log("Retrying")
        .setProperty("retrying", constant(true))
    .end()
    .log("Start")   
    .choice()
        .when(property("retrying").isNull())
            .log("Throwing")
            .throwException(new Exception("Hello world"))
        .end()
    .end()
    .log("Done")

显然这不是真正的路线；整个choice 正文只是在某些情况下模拟我的组件错误。我希望看到记录以下消息：

Start
Throwing
Retrying
Start
Done

但我实际看到的是：

Start
Throwing
Retrying
Failed delivery for (MessageId: ... on ExchangeId: ...). Exhausted after delivery attempt: 2 caught: java.lang.Exception: Hello world. Processed by failure processor: FatalFallbackErrorHandler[Pipeline[[Channel[Log(onExceptionTest)[Retrying]], Channel[setProperty(retrying, true)]]]]

我已经尝试将handled(true) 添加到异常处理程序中，但这所做的只是抑制错误消息。我没有看到第二条 Start 或 Done 日志消息。

为什么我的路线没有按照我的预期运行，我需要做什么才能让它按照我想要的方式运行？

更新

@ProgrammerDan 指出问题在于重新交付不是为了我想要实现的目标，这可以解释为什么我的路线不起作用！所以我需要在我的处理程序中完成这项工作，但是我的路由调用了一个 Web 服务并且还有一些其他步骤，我不想在处理程序中复制所有这些。我想出了这个，它按预期工作，但它涉及从一开始就再次调用自身的路由。这是一个坏主意吗？我会用这种方法让自己陷入困境吗？

from("direct:onExceptionTest")
    .onException(Exception.class)
        .onWhen(property("retrying").isNull()) // don't retry forever
        .log("Retrying")
        .setProperty("retrying", constant(true))
        .handled(true)
        .to("direct:onExceptionTest") // is recursion bad?
    .end()
    .log("Start")   
    .choice()
        .when(property("retrying").isNull())
            .log("Throwing")
            .throwException(new Exception("Hello world"))
        .end()
    .end()
    .log("Done")

Answer 1

使用onRedelivery 和Processor 来设置属性：

String KEY = "retrying";

from("direct:onExceptionTest")
     .onException(RuntimeException.class)
         .onRedelivery(new Processor() { // Sets a processor that should be processed before a redelivery attempt.
             @Override
             public void process(final Exchange exchange) throws Exception {
                 LOG.info("Retrying");
                 exchange.setProperty(KEY, true);
             }
        })
        .maximumRedeliveries(1)
        .handled(true)
    .end()
    .log("Start")
    .process(new Processor() {
        @Override
        public void process(final Exchange exchange) throws Exception {
            LOG.info("No problem");
        }
    })
    .process(new Processor() {
        @Override
        public void process(final Exchange exchange) throws Exception {
            if (exchange.getProperty(KEY) == null) {
                LOG.info("Throwing");
                throw new RuntimeException("Hello World");
            }
            else {
                LOG.info("No throwing");
            }
        }
    })
    .log("Done");

打印出来

[                          main] route1                         INFO  Start
[                          main] OnExceptionHandler             INFO  No problem
[                          main] OnExceptionHandler             INFO  Throwing
[                          main] OnExceptionHandler             INFO  Retrying
[                          main] OnExceptionHandler             INFO  No throwing
[                          main] route1                         INFO  Done

正如@ProgrammerDan 所指出的，只有失败的处理器会被重新执行，而不是第一个顺利通过且没有任何问题的处理器。

编辑：

如果必须重新完成所有处理，那么您可以使用带有doTry 和doCatch 的子路由，如下所示：

from("direct:onExceptionTest")
    .doTry()
        .to("direct:subroute")
    .doCatch(RuntimeException.class)
        .setProperty(KEY, constant(true))
        .to("direct:subroute")
    .end()
    .log("Done");

from("direct:subroute")
    .log("Start")
    .process(new Processor() {
        @Override
        public void process(final Exchange exchange) throws Exception {
            LOG.info("No problem");
        }
    })
    .process(new Processor() {
        @Override
        public void process(final Exchange exchange) throws Exception {
            if (exchange.getProperty(KEY) == null) {
                LOG.info("Throwing");
                throw new RuntimeException("Hello World");
            }
            else {
                LOG.info("No throwing");
            }
        }
    });

来自Camel Docs：

当使用 doTry .. doCatch .. doFinally 时，常规的 Camel 错误处理程序不适用。这意味着任何onException 或类似的东西都不会触发。原因是 doTry .. doCatch .. doFinally 实际上是它自己的错误处理程序，它旨在模仿和工作，就像在 Java 中的 try/catch/finally 一样。

Answer 2

关于 Camel 的重新投递机制需要考虑的几点。首先，查看the docs on the topic，这可能会挑战您对 Camel 如何处理重新交付的假设。我链接到的一点是骆驼尝试在故障点重新交付，它不会从路线的起点重新开始（正如您所假设的那样）。如果我对文档的理解正确（我有一段时间没有尝试过这种模式），你基本上是在告诉它多次重试抛出异常，我怀疑这是你想要测试的。

其次，我建议直接在onException() 处理器链中进行备用处理，如a little further down in the same docs 所示。基本上，您可以指定您希望如何通过自定义处理器处理消息，并同时使用handled(true) 和stop() 来表示不需要进一步处理。

总而言之，重新投递通常是为了处理典型的端点投递失败，例如间歇性连接中断、接收服务器暂时不可用等。在这种情况下，“再试一次”并有合理的成功预期是最有意义的.如果您需要更复杂的逻辑来处理重试，请在 onException() 处理器链中使用自定义处理器或一系列处理器。