如何基于Camel FTP路由按分组实现多文件FTP传输？

直接用Camel的Splitter EIP就能搞定，不需要手动在Process里拆分存Header/Property。Splitter本身可以把原始列表按指定规则拆成多个独立的Exchange，每个Exchange对应一组数据，天然支持多批次FTP传输。

具体实现步骤 #

1. 按attribute6拆分分组 #

用Splitter结合分组逻辑，把Body里的对象列表按attribute6分成多个子列表：

• Java DSL写法：
  先写个分组方法，把列表按attribute6分组后返回子列表的集合：

  public List<List<YourObject>> groupByAttr6(List<YourObject> items) {
      return new ArrayList<>(items.stream()
          .collect(Collectors.groupingBy(YourObject::getAttribute6))
          .values());
  }
  

  然后在路由里调用这个方法做拆分：

  from("direct:yourInputRoute")
      // 用GroupedExchangeAggregationStrategy保证每个子Exchange的Body是完整的分组列表
      .split(body(), new GroupedExchangeAggregationStrategy())
          .method(this, "groupByAttr6")
          // 后续处理每个分组
  

• XML路由写法：
  用OGNL表达式直接分组，不用额外写Java方法：

  <route>
      <from uri="direct:yourInputRoute"/>
      <split strategyRef="groupedExchangeAggregationStrategy">
          <simple>${body} groupBy attribute6</simple>
          <!-- 后续处理每个分组 -->
      </split>
  </route>
  <!-- 声明聚合策略 -->
  <bean id="groupedExchangeAggregationStrategy" class="org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy"/>
  

2. 为每个分组设置独立文件名 #

拆分后每个子Exchange对应一组数据，取组内第一个对象的attribute6值作为文件名的一部分，避免重名：

.process(exchange -> {
    List<YourObject> group = exchange.getIn().getBody(List.class);
    if (!group.isEmpty()) {
        // 自定义文件名，比如用attribute6值作为标识
        String fileName = "data_" + group.get(0).getAttribute6() + ".csv";
        exchange.getIn().setHeader("CamelFileName", fileName);
    }
})

XML配置的话，写个Processor类就行，逻辑和上面一致。

3. 执行FTP传输 #

在子路由里直接调用FTP端点，每个分组会独立发起一次传输：

.to("ftp://username:password@your-ftp-server/target-path?passiveMode=true")

完整Java DSL示例 #

from("direct:processAndFtp")
    // 按attribute6分组拆分
    .split(body(), new GroupedExchangeAggregationStrategy())
        .method(this, "groupByAttr6")
        // 设置FTP文件名
        .process(exchange -> {
            List<YourObject> group = exchange.getIn().getBody(List.class);
            if (!group.isEmpty()) {
                exchange.getIn().setHeader("CamelFileName", "group_" + group.get(0).getAttribute6() + ".json");
            }
        })
        // 把对象列表转成JSON（根据需求换成CSV等格式）
        .marshal().json(JsonLibrary.Jackson)
        // FTP传输
        .to("ftp://user:pass@ftp.example.com/upload?passiveMode=true")
    .end();

// 分组方法
public List<List<YourObject>> groupByAttr6(List<YourObject> items) {
    return new ArrayList<>(items.stream()
        .collect(Collectors.groupingBy(YourObject::getAttribute6))
        .values());
}

关键注意事项 #

• GroupedExchangeAggregationStrategy不能少：这个策略会让Splitter把每个完整的分组列表作为子Exchange的Body，而不是逐个传递列表里的对象。
• 文件名要唯一：必须用attribute6或者其他唯一标识做文件名，不然FTP服务器上的文件会被覆盖。
• 数据格式转换：如果要传输文本格式（比如CSV、JSON），一定要加marshal步骤，不然会直接传对象的toString结果，大概率不符合需求。

替代方案（不推荐，仅作参考） #

如果需要先聚合再拆分，可以先用Aggregate EIP按attribute6分组，再用Splitter拆分聚合后的结果，但这种方式比直接用Splitter分组复杂，没必要优先选用。

内容的提问来源于stack exchange，提问作者R M