只汇总具有相同关联密钥的连续交换

Question

我使用的是Apache Camel并获取一个大文件以进行输入，我必须逐行处理。内容已经排序，我必须使用相同的关联关键字聚合所有连续的行。如果相关键发生变化，则必须完成前一个聚合。如果文件结束，最后的聚合也已完成。 我有一些限制： - 由于传入文件相当大，我们希望以流式处理它。 - 因为结果赋予同步端点，所以我不想使用超时完成谓词。否则，我将失去调节数据源消耗速度的背压，并且交换将积累在AggregateProcessor的超时映射和聚合存储库中。

PreCompletionAwareAggregationStrategy看起来像一个有希望的解决方案，但事实证明，最后一个聚合将不会完成，直到下一个文件到达。如果我在preComplete中使用CamelSplitComplete属性，则最后一个聚合完成，但没有最后一个传入交换。相反，这最后的交换将被添加到下一个文件到达的内容。

因此，目前我很迷茫，找不到过分难看的解决方案。

Answer 1

在描述的场景，我会派劈裂消息的路由与聚合器（姑且称之为“AggregationRoute”），该公司聚集战略实现PreCompletionAwareAggregationStrategy（你已经在使用它的方式，我猜）。 然后，当分割结束时，将AGGREGATION_COMPLETE_ALL_GROUPS标头设置为true并将其发送到AggregationRoute。此交换仅用作完成所有聚合组的信号。

实施例：

    ... 
    .split(body()).streaming() 
     .to("direct:aggregationRoute") 
    .end() 
    .setHeader(Exchange.AGGREGATION_COMPLETE_ALL_GROUPS,constant(true)) 
    .to("direct:aggregationRoute"); 

from("direct:aggregationRoute") 
    .aggregate([your correlation expression]), myAggregationStrategy) 
    ... 

另一种替代方法是使用AggregateController通过调用其方法forceCompletionOfAllGroups（）来结束所有组的聚集：

AggregateController aggregateController = new DefaultAggregateController(); 

from(...) 
    ... 
    .split(body()).streaming() 
     .aggregate([correlation expression], aggregationStrategy).aggregateController(aggregateController) 
      ... 
      // Do what you need to do with the aggregated exchange 
      ... 
     .end() 
    .end() 
    .bean(aggregateController, "forceCompletionOfAllGroups") 

Answer 2

好吧，也许有一种方法可能是因为你的数据已经被排序了，要以流方式解析，并将每一行的相关键添加到一些hashmap结构中。一旦遇到新的关联关键字，你基本上想要“刷新”hashmap来创建一个新消息，然后重新启动相同的进程。看看这里： http://camel.apache.org/how-do-i-write-a-custom-processor-which-sends-multiple-messages.html