Apache Camel与Spring Boot中Split组件使用后无法序列化响应问题

算力阿峰

2026-7-6

问题描述

在Spring Boot集成Apache Camel的流程中，我执行了以下操作：

构造查询字符串列表存入Exchange Body，同时初始化空的ActivityIssuesDTO列表存入Exchange属性response
使用.split(body())拆分Body中的查询列表
每个拆分后的Exchange读取Body，将查询设置为HTTP_QUERY头，调用微服务并反序列化响应为ActivityIssuesDTO
将响应添加到response属性的列表中，最后尝试序列化该列表

遇到的问题：使用Split组件后，无法看到最终响应，也无法完成序列化；移除Split组件后，单条查询可正常获取响应并序列化，用静态数组测试也存在相同问题。

解决方案

问题根源

Exchange属性隔离：Split拆分出的子Exchange与父Exchange完全独立，子Exchange中对response属性的修改不会同步到父Exchange，最终父Exchange拿到的还是初始空列表。
Split默认无聚合：Split默认仅拆分任务并行/串行执行，不会自动收集子任务的返回结果，必须显式配置聚合策略才能合并所有子响应。

具体实现方案

1. 自定义聚合策略

创建AggregationStrategy实现类，专门收集子Exchange返回的ActivityIssuesDTO并合并为列表：

public class ActivityIssuesAggregationStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        // 第一次聚合：初始化结果列表
        if (oldExchange == null) {
            List<ActivityIssuesDTO> resultList = new ArrayList<>();
            resultList.add(newExchange.getIn().getBody(ActivityIssuesDTO.class));
            newExchange.getIn().setBody(resultList);
            return newExchange;
        }
        // 后续聚合：往已有列表添加新结果
        List<ActivityIssuesDTO> existingList = oldExchange.getIn().getBody(List.class);
        existingList.add(newExchange.getIn().getBody(ActivityIssuesDTO.class));
        oldExchange.getIn().setBody(existingList);
        return oldExchange;
    }
}

2. 调整Camel路由配置

在Split组件中指定聚合策略，确保所有子任务执行完成后，父Exchange的Body是合并后的完整列表，再执行序列化：

from("direct:start")
    // 构造查询字符串列表（示例逻辑，替换为你的实际构造代码）
    .process(exchange -> {
        List<String> queryList = Arrays.asList("status=open", "priority=high", "assignee=me");
        exchange.getIn().setBody(queryList);
    })
    // 拆分列表并绑定聚合策略，parallelProcessing为可选配置
    .split(body(), new ActivityIssuesAggregationStrategy())
        .parallelProcessing() // 可选：开启并行处理，提升效率
        .process(exchange -> {
            String query = exchange.getIn().getBody(String.class);
            exchange.getIn().setHeader(Exchange.HTTP_QUERY, query);
        })
        // 调用目标微服务，替换为你的实际服务地址
        .to("http://your-microservice-api/issues")
        // 反序列化响应为ActivityIssuesDTO
        .unmarshal().json(JsonLibrary.Jackson, ActivityIssuesDTO.class)
    .end() // 结束Split，回到父Exchange上下文
    // 序列化最终结果列表
    .marshal().json(JsonLibrary.Jackson)
    .to("log:final-aggregated-response");

3. 关键注意事项

并行处理时，ArrayList不是线程安全的，可替换为CopyOnWriteArrayList或在聚合逻辑中加锁保证线程安全。
确保单个子请求的反序列化逻辑正常，否则聚合时会出现类型转换异常。
不要依赖Exchange属性传递聚合结果，Split的子Exchange无法修改父Exchange的属性，必须通过AggregationStrategy完成结果合并。

内容的提问来源于stack exchange，提问作者Sayed Sajad Hosseini