使用Apache Camel和ActiveMQ进行错误处理-因此突破管道进行交换

Question

我一直在反复讨论我们系统中的一个问题，即使在论坛上进行了一些研究和多次测试，我们似乎也无法解决这个问题。

我会尽量弄清楚我们在处理什么

我们有一个带有路由的主服务，该路由从activemq队列读取数据（使用嵌入式代理的spring boot）将其发送到路由（B），然后将所有内容发送到最终路由（C）。路由（B）依赖于服务。

骆驼版：3.3.0Spring靴版：2.3.3。发布

路线A：

 onException(Exception::class.java)
            .handled(true)
            .bean("foo.ErrorProcessor", "processError")

 from("activemq:queue:myqueue")
            .routeId("myroute")
            .to("direct:my_external_route")
            .to(ExchangePattern.InOnly,"direct:myroute_result")

路线B：

 onException(Exception::class.java)
            .handled(true)
            .bean("foo.ErrorProcessor", "processError")

 from("direct:my_external_route")
            .routeId("my_external_route")
            .process {something()} //This processor can throw exceptions that are treated in our processor

路线C：

from("direct:myroute_result")
      .process(someProcess())
      .to(ExchangePattern.InOnly,"activemq:queue:results_queue")

Spring Boot activemq配置

spring:
  jmx:
    enabled: true
  activemq:
    broker-url: vm://localhost?broker.persistent=false,useShutdownHook=false
    in-memory: true
    non-blocking-redelivery: true
    packages:
      trust-all: false
      trusted: com.mypackage
    pool:
      block-if-full: true
      block-if-full-timeout: -1
      enabled: false
      idle-timeout: 30000
      max-connections: 10
      time-between-expiration-check: -1
      use-anonymous-producers: true

当B的处理器不抛出异常时，一切都运行得很好、顺利。当它发生时，即使它们正在被处理，并且在消息体中返回了一个普通对象，我们在日志中的所有内容都是

2021-04-10 15:33:32.354 DEBUG [#1 - JmsConsumer[consumerName]] o.a.c.p.Pipeline                   
: Message exchange has failed: so breaking out of pipeline for exchange: Exchange[ID-1234] Handled by the error handler. {}

我们甚至在activemq连接工厂中添加了一个默认错误处理程序，但那里也没有发生任何事情。我们有一个DLQ消费者，他似乎什么也得不到。routeA上的错误处理器也不会捕获任何预期的内容，因为之前处理过异常。

有没有人有过这个问题或类似的问题？我知道驼峰和JMS组件之间在错误处理方面存在一些问题，但我们很难理解这个问题的根源是什么。

先谢谢你，佩德罗

Answer 1

可能您正在寻找的是路线B例外条款中的继续选项。此选项允许您继续路由到原始路由，就像没有发生异常一样。不要使用handled选项，因为它将不允许路由到原始路由，但会中断。因此，路线B的定义如下：

onException（Exception::class.java）
.继续（true）
. bean（"foo. ErrorProcess"，"Process Error"）

来自（“direct:my\u external\u route”）
。routeId（“my\u external\u route”）
。进程{something（）}有关更多详细信息，请参阅camel文档：camel EXCEPTION子句