按Id进行groupby分组拆分大平面文件，批量存储并对接Camel File组件

步骤1：完成Id分组后的批量打包处理 #

假设你已经通过Java 8 Stream完成按Id分组，得到Map<String, List<String>> idGroups（键为Id值，值为对应行的列表，已跳过表头）。接下来将这些分组按指定数量（比如2个）打包成批次：

// 已完成的按Id分组结果
Map<String, List<String>> idGroups = ...;

// 将分组条目转为列表，方便按批次拆分
List<Map.Entry<String, List<String>>> groupEntries = new ArrayList<>(idGroups.entrySet());

// 定义每个文件包含的分组数
int groupsPerFile = 2;
List<List<Map.Entry<String, List<String>>>> batchedGroups = new ArrayList<>();

for (int i = 0; i < groupEntries.size(); i += groupsPerFile) {
    int endIndex = Math.min(i + groupsPerFile, groupEntries.size());
    batchedGroups.add(groupEntries.subList(i, endIndex));
}

步骤2：生成每个批次对应的文件内容 #

遍历每个批次，合并该批次下所有分组的行内容：

List<String> fileContents = batchedGroups.stream()
    .map(batch -> batch.stream()
        .flatMap(entry -> entry.getValue().stream())
        .collect(Collectors.joining(System.lineSeparator())))
    .collect(Collectors.toList());

如果需要给每个文件添加表头，可修改映射逻辑：

String header = "Name Id";
List<String> fileContentsWithHeader = batchedGroups.stream()
    .map(batch -> {
        List<String> lines = batch.stream()
            .flatMap(entry -> entry.getValue().stream())
            .collect(Collectors.toList());
        lines.add(0, header);
        return String.join(System.lineSeparator(), lines);
    })
    .collect(Collectors.toList());

步骤3：通过Camel File组件发送拆分后的文件 #

方式1：直接使用ProducerTemplate循环发送 #

ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
for (int i = 0; i < fileContents.size(); i++) {
    producerTemplate.sendBodyAndHeader(
        "file:/path/to/your/target/directory",
        fileContents.get(i),
        Exchange.FILE_NAME,
        "file" + (i + 1)
    );
}

方式2：通过Camel路由批量处理 #

// 定义路由
from("direct:dispatchSplitFiles")
    .process(exchange -> {
        List<String> contents = exchange.getIn().getBody(List.class);
        CamelContext context = exchange.getContext();
        for (int i = 0; i < contents.size(); i++) {
            Exchange subExchange = context.createExchange();
            subExchange.getIn().setBody(contents.get(i));
            subExchange.getIn().setHeader(Exchange.FILE_NAME, "file" + (i + 1));
            context.createProducerTemplate().send("file:/path/to/your/target/directory", subExchange);
        }
    });

// 触发路由
producerTemplate.sendBody("direct:dispatchSplitFiles", fileContents);

完整整合示例 #

// 1. 读取原始文件并按Id分组（跳过表头）
Map<String, List<String>> idGroups = Files.lines(Paths.get("/path/to/input/your-file.txt"))
    .skip(1)
    .collect(Collectors.groupingBy(line -> line.split("\\s+")[1]));

// 2. 按指定分组数打包批次
int groupsPerFile = 2;
List<Map.Entry<String, List<String>>> groupEntries = new ArrayList<>(idGroups.entrySet());
List<List<Map.Entry<String, List<String>>>> batchedGroups = new ArrayList<>();
for (int i = 0; i < groupEntries.size(); i += groupsPerFile) {
    batchedGroups.add(groupEntries.subList(i, Math.min(i + groupsPerFile, groupEntries.size())));
}

// 3. 生成带表头的文件内容
String header = "Name Id";
List<String> fileContents = batchedGroups.stream()
    .map(batch -> {
        List<String> lines = batch.stream()
            .flatMap(entry -> entry.getValue().stream())
            .collect(Collectors.toList());
        lines.add(0, header);
        return String.join(System.lineSeparator(), lines);
    })
    .collect(Collectors.toList());

// 4. 发送至Camel File组件
ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
for (int i = 0; i < fileContents.size(); i++) {
    producerTemplate.sendBodyAndHeader(
        "file:/path/to/output/directory",
        fileContents.get(i),
        Exchange.FILE_NAME,
        "file" + (i + 1)
    );
}

注意事项 #

• 处理超大文件时，建议结合Camel的流式拆分与分组，避免内存溢出。
• 可根据需求灵活调整groupsPerFile参数，修改每个文件包含的分组数量。

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