Apache Camel并行处理多播未将异常传递至死信处理器

这个问题我之前处理过，当你在Camel的Multicast组件中启用parallelProcessing()时，默认情况下并行分支抛出的异常不会自动传递到主路由的死信处理器——因为并行线程的执行上下文和主路由是相互独立的，异常被局限在各自的并行线程里，没法冒泡回主路由的错误处理链路。下面给你几个靠谱的解决方案：

方案1：启用stopOnException() #

这是最直接的解决方式，给Multicast添加stopOnException()配置后，只要任意一个并行分支抛出异常，Camel会立即终止所有其他并行分支的执行，并且把异常传递回主路由的死信处理器。修改你的路由代码如下：

from(errorMultiDirect).routeId("errorMulticastTest") 
    .errorHandler(deadLetterChannel(mock) 
        .onPrepareFailure(errorProcessor).maximumRedeliveries(0)) 
    .log(LoggingLevel.INFO, "Testing Error route") 
    .setHeader(OrderMessageConstants.WIMS_MSG_TYPE, simple("body[messageType]")) 
    .setHeader(OrderMessageConstants.SAP_MESSAGE_ID, simple("body[messageID]")) 
    .setHeader(OrderMessageConstants.ORDER_NUMBER, simple("body[orderHeader][order]")) 
    .multicast()
        .parallelProcessing()
        .stopOnException() // 新增这个配置
        .shareUnitOfWork()
        // 这里添加你的Multicast目标端点
        .to("direct:endpoint1", "direct:endpoint2")
    .end();

方案2：为并行分支单独配置错误处理 #

如果你的业务需求不允许因为单个分支异常就终止所有并行任务，可以给每个并行分支单独设置错误处理逻辑，手动将异常路由到死信通道。比如在分支路由里添加onException()：

// 定义分支路由1
from("direct:endpoint1")
    .onException(Exception.class)
        .handled(true)
        .to(mock); // 直接发送到死信端点
    .end()
    // 分支1的业务逻辑
    .log("Processing endpoint1");

// 定义分支路由2
from("direct:endpoint2")
    .onException(Exception.class)
        .handled(true)
        .to(mock);
    .end()
    // 分支2的业务逻辑
    .log("Processing endpoint2");

方案3：结合shareUnitOfWork()和自定义线程池 #

如果需要更精细的线程控制，你可以自定义ExecutorService并配置给Multicast，同时确保shareUnitOfWork()已启用——这样可以让并行线程共享主路由的UnitOfWork上下文，异常更容易被主路由的错误处理器捕获：

// 自定义线程池
ExecutorService customExecutor = Executors.newFixedThreadPool(5);

// 修改主路由
from(errorMultiDirect).routeId("errorMulticastTest") 
    .errorHandler(deadLetterChannel(mock) 
        .onPrepareFailure(errorProcessor).maximumRedeliveries(0)) 
    .log(LoggingLevel.INFO, "Testing Error route") 
    .setHeader(OrderMessageConstants.WIMS_MSG_TYPE, simple("body[messageType]")) 
    .setHeader(OrderMessageConstants.SAP_MESSAGE_ID, simple("body[messageID]")) 
    .setHeader(OrderMessageConstants.ORDER_NUMBER, simple("body[orderHeader][order]")) 
    .multicast()
        .parallelProcessing()
        .shareUnitOfWork()
        .executorService(customExecutor) // 配置自定义线程池
        .to("direct:endpoint1", "direct:endpoint2")
    .end();

总结一下，方案1是最推荐的通用解决方案，如果有特殊业务需求再考虑方案2或3。

内容的提问来源于stack exchange，提问作者whomer