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

首先有一个类似的未回答的问题将路由加入单个聚合器

我们有一些消费者路由（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 个包含来自 ftp 的 6 个消息。

测试用例：

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，那么超时将从第一次交换开始发生——如果仍然有新的交换，则独立？

你几乎已经可以工作了。这是您需要的更改（之后我将解释）。

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.

解释一下 - 正如您所读到的，聚合器是一个多功能的骆驼组件，正确定义的关键是：

• 聚合表达式
• 完成规则

现在，就您而言，您正在聚合一处房产AGGREGATION_PROPERTY这将是A, B or Z。此外，您还要指定批量大小。

然而你并没有表达completionSize()在你的路线中。相反，你正在使用completionFromBatchConsumer- 它做了一些不同的事情（代码指出它寻找一个Exchange#BATCH_COMPLETE属性），因此产生了奇怪的结果。

Anyway, .completionSize(Exchange.BATCH_SIZE)将使您的测试按预期运行。

祝你好运。