Camel拆分后,Exception不会升级

Question

我们在Camel中定义了一个路由,并且如果在处理器中抛出异常,则必须找出它.当我们只有一个处理器时,Camel会在sendBody()方法中重新抛出异常.如果存在先前的分割/聚合,则不会抛出异常.所以下面例子的结果是

before throwing Exception

after sendBody

如果我省略从.split到.completionSize(1)的所有内容,则输出为

before throwing Exception

Exception thrown

任何想法如何找出,如果分裂后发生异常？

private static final String DIRECT_START = "direct:start";

public static void main(String[] args) throws Exception {
    CamelContext context = new DefaultCamelContext();

    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {

            from(DIRECT_START)
            .split(body())
                .aggregate(constant(true), new AggregationStrategy() {
                    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                        return oldExchange == null ? newExchange : oldExchange;
                    }
                })
                .completionSize(1)
            .process(new Processor() {
                public void process(Exchange exchange) throws Exception {
                    System.out.println("before throwing Exception");
                    exchange.setException(new Exception());
                    throw new Exception("my Exception");
                }
            });
        }});        

    context.start();

    ProducerTemplate producer = context.createProducerTemplate();
    try {
        producer.sendBody(DIRECT_START, Integer.valueOf(42));
        System.out.println("after sendBody");
    } catch (Exception e) {
        System.out.println("Exception thrown");
    }

    context.stop();

}

为了检查之后的异常,我们找到了一个解决方案.我们在onException()中注册了一个ErrorProcessor,它将状态设置为上下文属性.

但这不会中断producer.sendBody(..).我们有极端长时间运行的处理器,我们必须打断它们.

所以问题是,我们可以配置Camel在sendBody中抛出异常,还是可以在Exceptionhandler中执行此操作？

Answer 1

关于Splitter EIP和Camel in Action中的异常处理(第8.3.5节),我强烈推荐这一章.该节解释说:

将自定义AggregationStrategy与Splitter一起使用时,了解您负责处理异常非常重要.如果不传回异常,Splitter将假定您已处理异常,并忽略它.

您已使用该split()方法而未指定聚合器.在Camel文档中,他们指定了

默认情况下,拆分器将返回原始输入消息

这意味着离开split()方法的交换没有异常,因此没有异常传播回您的调用代码.从处理器抛出的异常在技术上是在分离器内部.即使您使用了聚合器,它也与split调用没有关联,并且您没有明确地结束split使用end().因此,当您的处理器抛出异常时,拆分器会忽略它,因为您没有提供聚合器来处理和传播异常.

我们可以通过将聚合策略split作为参数传递给调用来测试这一点,如下所示:

.split(body(), new AggregationStrategy() {
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        System.out.println("Aggregating");
        return oldExchange == null ? newExchange : oldExchange;
    }
})
    .log("test") // inside the split/aggregator EIP
.end() // outside the split/aggregator EIP
.process(new Processor() {
    public void process(Exchange exchange) throws Exception {
        System.out.println("before throwing Exception");
        throw new Exception("my Exception");
    }
});

你会得到输出:

test
Aggregating
before throwing Exception
Exception thrown

如果您希望处理器位于拆分/聚合器EIP内,如下所示:

.split(body(), new AggregationStrategy() {
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        System.out.println("Aggregating");
        return oldExchange == null ? newExchange : oldExchange;
    }
})
    .process(new Processor() { // inside the split/aggregator EIP
        public void process(Exchange exchange) throws Exception {
            System.out.println("before throwing Exception");
            throw new Exception("my Exception");
        }
    })
.end(); // outside the split/aggregator EIP

你会得到输出:

before throwing Exception
Aggregating
Exception thrown

请注意,在分割/聚合器EIP内部,如何在抛出异常后运行聚合器？这很重要,因为没有聚合器传递异常,分割器将忽略它.为了使其工作,您需要在聚合器内正确传播异常.例如,在您的代码中,如果newExchange要包含异常,那么它将被忽略,因为您没有传播它.您需要更改要添加的聚合器:

if (newExchange.getException() != null) {
    oldExchange.setException(newExchange.getException());
}

注意:如果您onException()在拆分EIP内部有一个呼叫,并且您将异常设置为正在处理,则在您呼叫时它将不再返回getException().因此,如果您想处理异常,但仍然通过聚合器传播它们,则可以使用exchange.getProperty(Exchange.EXCEPTION_CAUGHT);

您也可以使用.stopOnException(),如下所示:

.split(body()).stopOnException()
    .process(new Processor() { // inside the split/aggregator EIP
        public void process(Exchange exchange) throws Exception {
            System.out.println("before throwing Exception");
            throw new Exception("my Exception");
        }
    });

这会导致拆分在异常上停止,并传播它.但是,当您在聚合器之后放置聚合器时stopOnException(),它将不再起作用.我不完全确定为什么.我猜它是因为聚合器改变了exchange对象.

另请注意,您无需在处理器中将异常设置为交换.当处理器抛出异常时,Camel会为您执行此操作.因此,exchange.setException(new Exception());处理器中的线路不是必需的.

tl; dr是的,您可以从分割内部向调用方法传播异常.您只需确保通过与拆分或集合关联的聚合器完成它stopOnException().这取决于你试图通过拆分/聚合/处理最佳方法来实现的目标.