Camel 2.11批处理聚合如何与单独的路由一起工作？

Question

我们有一些使用方路径（ftp，file，smb）从远程系统读取文件。简化了直接路径测试，但与批处理使用者的行为类似：

from("direct:"+routeId).id(routeId)
 .setProperty(AGGREGATION_PROPERTY, constant(routeId))
 .log(String.format("Sending (${body}) to %s", "direct:start1"))
 .to("direct:aggregate");

转换后，一次轮询的所有结果将在单独的路径中按批汇总：

from("direct:aggregate")
  .aggregate(property(AGGREGATION_PROPERTY), new BodyInAggregatingStrategy())
  .completionFromBatchConsumer()
  .to("log:result", "mock:result");

如果每个消费者分开运行，则一切正常。但是，如果多个使用者并行运行，聚合将拆分民意测验。例如，如果文件消费者轮询500条消息，并且第二条路线开始从ftp读取6个文件，则期望得到2个聚合1，其中包含来自文件的500条消息，以及1个聚合，其中6条来自ftp的消息。

测试用例：

public void testAggregateByProperty() throws Exception {
    MockEndpoint result =  getMockEndpoint("mock:result");

    result.expectedBodiesReceived("A+A+A", "B+B", "A", "Z");

    template.sendBodyAndProperty("direct:A", "A", Exchange.BATCH_SIZE, 3);
    template.sendBodyAndProperty("direct:A", "A", Exchange.BATCH_SIZE, 3);
    template.sendBodyAndProperty("direct:B", "B", Exchange.BATCH_SIZE, 2);
    template.sendBodyAndProperty("direct:A", "A", Exchange.BATCH_SIZE, 3);
    template.sendBodyAndProperty("direct:B", "B", Exchange.BATCH_SIZE, 2);
    template.sendBodyAndProperty("direct:A", "A", Exchange.BATCH_SIZE, 1);
    template.sendBodyAndProperty("direct:Z", "Z", Exchange.BATCH_SIZE, 7);

    assertMockEndpointsSatisfied();
}

结果是：“ A + A”，“ B”，“ A”，“ B”，“ A”，而不是预期的“ A + A + A”，“ B + B”，“ A”，“ Z” 。问题：

1. 我们对聚合的假设是否错误？
2. 我们如何实现预期的行为？
3. 如果我们设置了completionTimeout，则表明第一次交换将发生超时-如果仍然有新的交换，则独立吗？

Answer 1

您几乎可以正常工作了。这是您需要的更改（稍后我会解释）。

from("direct:aggregate").id("aggregate")
    .aggregate(property(AGGREGATION_PROPERTY), new BodyInAggregationStrategy())
    .completionSize(property(Exchange.BATCH_SIZE))
    .to("log:result", "mock:result")

结果将是：

Exchange received, body: A+A+A
Exchange received, body: B+B
Exchange received, body: A

注意："Z"由于批次大小为，您将不会收到结果7。

解释一下-如您所读， Aggregator 是一个通用的骆驼组件，正确定义的关键是：

• 聚合表达式
• 完成规则

现在，在你的情况你是聚合上的属性AGGREGATION_PROPERTY，这将是A，B或Z。另外，您要指定批次大小。

但是，您并未completionSize()在路线中表达。取而代之的是，您使用了completionFromBatchConsumer-做一些不同的事情（代码指出它在寻找Exchange#BATCH_COMPLETE属性），因此结果很奇怪。

无论如何，.completionSize(Exchange.BATCH_SIZE)将使您的测试按需运行。

祝你好运。