Apache Camel：split+streaming+aggregate后enrich()执行顺序异常及解决

问题根源 #

当Split开启streaming模式时，Camel会采用异步非阻塞的方式处理分片数据，默认情况下Split-Aggregate流程不会阻塞主线程等待所有分片处理完成并聚合完毕，导致Enrich调用Route2的同时，Route1后续逻辑（比如日志）提前执行，违背了“Route2完全执行完再触发Enrich聚合策略”的预期。

可行解决方案（保留Split+Streaming+Aggregate逻辑） #

方案1：强制Split-Aggregate流程同步阻塞 #

在Split和Aggregate环节添加配置，确保主线程等待整个分片处理与聚合流程完成后，再执行后续的Enrich逻辑：

• 显式禁用Split的并行处理（默认已禁用，显式声明更稳妥）
• 为Aggregate指定明确的完成条件（比如根据分片总数）
• 添加waitForCompletion()强制等待异步任务完成

代码示例（Java DSL）：

// Route1 配置
from("direct:route-1")
    .split(body()).streaming()
        .parallelProcessing(false) // 禁用并行，确保分片顺序处理
        .process(exchange -> {
            // 分片处理逻辑
        })
    .aggregate(constant(true), new Route1AggregationStrategy())
        .completionSize(header(Exchange.SPLIT_SIZE)) // 所有分片处理完成后触发聚合
        .waitForCompletion() // 等待聚合完成，阻塞主线程
    .enrich("direct:route-2", new EnrichAggregationStrategy())
    .log("route-1 执行完成");

// Route2 配置
from("direct:route-2")
    .split(body()).streaming()
        .parallelProcessing(false)
        .process(exchange -> {
            // Route2分片处理逻辑
        })
    .aggregate(constant(true), new Route2AggregationStrategy())
        .completionSize(header(Exchange.SPLIT_SIZE))
        .waitForCompletion()
    .log("route-2 执行完成");

XML配置示例：

<!-- Route1 -->
<route id="route-1">
    <from uri="direct:route-1"/>
    <split streaming="true" parallelProcessing="false">
        <simple>${body}</simple>
        <process ref="route1SplitProcessor"/>
    </split>
    <aggregate completionSize="${header.CamelSplitSize}" waitForCompletion="true">
        <correlationExpression>
            <constant>true</constant>
        </correlationExpression>
        <aggregationStrategy ref="route1AggregationStrategy"/>
    </aggregate>
    <enrich uri="direct:route-2" aggregationStrategyRef="enrichAggregationStrategy"/>
    <log message="route-1 执行完成"/>
</route>

<!-- Route2 -->
<route id="route-2">
    <from uri="direct:route-2"/>
    <split streaming="true" parallelProcessing="false">
        <simple>${body}</simple>
        <process ref="route2SplitProcessor"/>
    </split>
    <aggregate completionSize="${header.CamelSplitSize}" waitForCompletion="true">
        <correlationExpression>
            <constant>true</constant>
        </correlationExpression>
        <aggregationStrategy ref="route2AggregationStrategy"/>
    </aggregate>
    <log message="route-2 执行完成"/>
</route>

方案2：调整Aggregate的完成策略 #

如果必须保留异步处理，可以通过更精准的完成条件确保聚合完成后再触发Enrich：

• 使用completionPredicate自定义完成判断逻辑，比如检查所有分片的处理状态
• 结合completionTimeout设置超时时间，避免无限等待（适合数据量波动大的场景）

关键注意点 #

• waitForCompletion()是核心：它会让Camel等待Split-Aggregate的所有异步任务结束，确保主线程不会提前执行后续逻辑
• 完成条件必须准确：completionSize依赖Exchange.SPLIT_SIZE头（Split自动设置），确保聚合在所有分片处理完成后触发
• Enrich默认是同步调用：只要Route2的Split-Aggregate流程完成，Enrich的聚合策略就会在Route2执行完毕后触发

内容的提问来源于stack exchange，提问作者T.Rex