Camel - 在拆分中处理异常并继续

Question

在我的骆驼路线中，我有一个像这样的简单分割

                .split(body(), eventListAggregationStrategy).parallelProcessing()
                    .process(rawEventTransformationProcessor)
                .end()

如果.process(rawEventTransformationProcessor) 内部发生异常，我希望它们被处理并发送到错误队列。然而，我所有的尝试都失败了，整条路线都停止了。 我以多种方式尝试了 onException（使用handled、continue、shareUnitOfWork）等等。

        onException(RawEventTransformationException.class)
                .to("log:RawEventTransformationException?showAll=true&multiline=true&level=ERROR")
                .handled(true)
                .process(exchange -> {
                    RawEventTransformationException cause = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, RawEventTransformationException.class);
                    exchange.getIn().setBody(cause.getFailedEvent());
                })
                .marshal().json(JsonLibrary.Jackson, RawEvent.class)
                .to("rabbitmq://errors?queue=transformed-data-fails&routingKey=transformed-data-fails&autoDelete=false");

如果我有一个额外的direct 路由，当异常起作用时，我会从处理器内部触发该路由。

catch (RawEventTransformationException e1) {
            producerTemplate.sendBody(e1);
            exchange.getIn().setBody(List.of());
            throw e1;
        }

捕获此异常并让其余异常继续的最佳做法是什么？

Answer 1

使用continued(true) 的路由级异常处理至少对我来说效果很好。所有消息拆分消息都得到处理并处理异常。结帐Dead Letter channel 以获取有关将失败的交换发送到错误队列的信息。

您还可以在拆分块内调用直接路由，并在那里使用路由级异常处理来将异常处理限制为仅拆分消息。

public class ExampleTest extends CamelTestSupport {
    
    @Test
    public void exampleTest() {

        List<String> body = Arrays.asList(new String[]{"a", "b", "exception", "c", "d"});
        template.sendBody("direct:start", body);
    }

    @Override
    protected RoutesBuilder createRouteBuilder() throws Exception {
        
        return new RouteBuilder() {

            @Override
            public void configure() throws Exception {
             
                from("direct:start")
                    .onException(Exception.class)
                        .log(LoggingLevel.ERROR, "Exception handled: ${exception.message}")
                        .continued(true)
                    .end()
                    .log(LoggingLevel.INFO, "Start")
                    .split(body()).parallelProcessing()
                        .choice()
                            .when(body().isNotEqualTo("exception"))
                                .log(LoggingLevel.INFO, "${body}")
                            .otherwise()
                                .process(exchange -> {
                                    throw new Exception("Example exception");
                                })
                        .end()
                    .end()
                    .log(LoggingLevel.INFO, "Done");
            }
        };
    }
}

日志：

INFO  [log]         org.apache.camel.spi.CamelLogger : Start
INFO  [log]         org.apache.camel.spi.CamelLogger : b
INFO  [log]         org.apache.camel.spi.CamelLogger : a
INFO  [log]         org.apache.camel.spi.CamelLogger : c
INFO  [log]         org.apache.camel.spi.CamelLogger : d
ERROR [log]         org.apache.camel.spi.CamelLogger : Exception handled: Example exception
INFO  [log]         org.apache.camel.spi.CamelLogger : Done

Answer 2

您可以让 EventListAggregationStrategy 在它返回的交换中清除异常：

@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
    ...
    newExchange.setException(null);
    return newExchange;
}