我们在Camel中定义了一个具有拆分和聚合功能的路由,但是不能将异常传播回聚合器之后的拆分。
下面是这是不工作的代码
from("direct:MyRoute")
.routeId("MyRouteID")
.split().tokenize("\n", 1)
.streaming().stopOnException()
.choice()
.when(simple("${property.CamelSplitIndex} > 0"))
.unmarshal(domainDataFormat)
.choice()
.when(simple("${property.CamelSplitComplete}"))
.process(
new Processor()
{
@Override
public void process(Exchange exchange) throws Exception
{
exchange.getIn().getHeaders().put(Exchange.AGGREGATION_COMPLETE_ALL_GROUPS_INCLUSIVE, true);
}
}
)
.end()
.aggregate(myAggregationStrategy).constant(true) //if i comment this line split will be stop on exception
.threads().executorService(executorService)
.process(myProcessor).end()
.end();
处理器(myProcessor)从上面的代码如下:
counter.incrementAndGet(); //atomic counter
if(counter.get()==3)
{
exchange.setException(new RuntimeException());
throw new RuntimeCamelException();
}
但是,当我从路由中删除聚合时,Split能够在出现异常时停止路由。
1条答案
按热度按时间gwbalxhn1#
使用复合消息处理器EIP,它是同一工作单元中的拆分+聚合(fork/join)。
参见以下网址的文档:https://camel.apache.org/components/next/eips/composed-message-processor.html
并参见 splitter only 部分,在该部分中,您可以为拆分器指定聚合策略以使其协同工作。