分割EIP中的骆驼异常路由

Question

我正在将CSV文件输入集成到一个更大的系统中，我想对其进行设置，以便将CSV中未正确解析的各行（由split）发送到医院队列，并将所有其他工作行转换为一个对象并聚合到一个列表中。但是，我很难将拆分中抛出的异常消息路由到其他位置，而不会出现在拆分结束时的聚合器中。为了简化它，我编写了一个单元测试，在分割中抛出一个异常，并尝试让它工作。我将字符串1\n2\n3\n拆分为三条消息，在其中一条消息上引发异常，并将其余字符串连接到末尾。

public void configure() throws Exception {

    onException(Exception.class)
        .handled(true)
        .to("log:dead?level=ERROR");

    from("direct:test")
        .split(body().tokenize("\n"), new MyAggregationStrategy())
            .process(new ThrowMyException())
        .end()
        .to("mock:out");
}

@Test
public void test() throws InterruptedException {

    MockEndpoint out = getMockEndpoint("mock:out");
    out.expectedMessageCount(1);

    template.sendBody("direct:test", "1\n2\n3\n");

    assertMockEndpointsSatisfied();
}

ThrowMyException控制器：只是抛出一个异常

@Override
public void process(Exchange exchange) throws Exception {
    System.out.println("[" + exchange.getIn().getBody() + "]");
    if (exchange.getIn().getBody(String.class).trim().contentEquals("2")) {
        System.out.println("Throwing exception");
        throw new Exception();
    }
}

MyAggregationStrategy：只是连接字符串

@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
    if (oldExchange == null) {
        return newExchange;
    }
    oldExchange.getIn().setBody(
            oldExchange.getIn().getBody(String.class) + 
            newExchange.getIn().getBody(String.class));
    System.out.println(oldExchange.getIn().getBody());
    return oldExchange;
}

我所期望的是，onException处理程序，正如我所指出的，它将处理异常（handled（true）），它将在收到消息“2”时使用控制器抛出的异常，我将得到聚合器打印出的结果13。我确实得到了由onException处理程序打印出来的行：

错误死-交换[ExchangePattern:InOnly，BodyType:String，Body:2]

但聚合器随后返回“123”而不是“13”。

我也尝试过使用死信频道

errorHandler(deadLetterChannel("log:dead?level=ERROR")

我还尝试在onException处理程序中指定continued（false），但没有成功。

我意识到我可以通过查看exchange在聚合器中查找异常。getProperty（Exchange.EXCEPTION_capture），然后不聚合它。所以我有一个解决办法。但是，如果有一种方法可以在我的路线中使用onException或errorHandler，我更愿意这样做。




Answer 1

骆驼文档为停止OnExc0019分配器选项说

是否在发生异常时立即停止或继续处理。如果禁用，则继续分割并处理子消息，而不管其中一个消息是否失败。您可以在AggregationStrategy类中处理异常，您可以完全控制如何处理异常。

这以及Camel的作用表明，两个拆分器异常处理选项是在聚合器中处理异常，或者使用stopOnException（），这将阻止整个拆分器迭代器继续运行。

您可以使用exchange.get异常（）方法，或exchange.is失败（）来测试聚合器中的异常和错误。如果您使用了处理（true）上的onExc0019方法，则必须使用exchange.get属性（Exchange.EXCEPTION_CAUGHT）