使用Spring Integration批量发送数据如何实现GCP大文件分批读取路由？

核心疑问解答 #

首先可以明确：Spring Integration 的 Splitter 原生支持惰性流处理，不会等待整个流加载完成才输出数据，完全可以实现和你当前手动分批效果一致的能力，且无需加载全量文件数据到内存。

具体实现逻辑说明 #

• Splitter 对 Stream、Iterator 这类惰性迭代类型的 payload 会做逐元素迭代处理，每读取到流中的一个元素就会向下游发送，不会提前将全量流元素加载到内存。你只需要将现有的 file-reader 改造为返回反序列化后的对象流 Stream<S>，而非全量 Collection 即可，无需自己维护批次计数、临时集合缓存逻辑。
• 你之前担心的「加 aggregator 就要返回全量集合」是误区：只需给 aggregator 配置基于消息数量的释放策略，即可实现按固定批次大小输出，不会攒全量数据：
  • 配置 releaseStrategy 为 MessageCountReleaseStrategy(BATCH_SIZE)，攒够指定数量的消息就立即释放当前批次
  • 配置 expireGroupsUponCompletion = true，释放批次后立即清空对应分组缓存，不会留存历史数据
  • 可选配置 groupTimeout，处理最后不足批次大小的剩余数据，超时后自动释放，和你原有代码最后返回剩余临时集合的效果一致

改造后的代码示例 #

1. 流式读取方法改造

public <S> Stream<S> readDataStream(DataInfo dataInfo, Class<S> clazz) {
    try {
        Resource gcpResource = context.getResource("classpath://data.json");
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(gcpResource.getInputStream()));
        return bufferedReader.lines()
                .map(dataStr -> {
                    try {
                        return deserializeData(dataStr, clazz);
                    } catch (JsonProcessingException ex) {
                        throw new CustomException("PARSER-1001", "Error occurred while parsing", ex);
                    }
                })
                .onClose(() -> {
                    try {
                        bufferedReader.close();
                    } catch (IOException e) {
                        throw new CustomException("PARSER-1002", "Error closing file stream", e);
                    }
                });
    } catch (IOException ex) {
        throw new CustomException("PARSER-1002", "Error occurred while reading file", ex);
    }
}

2. 集成流配置（Java DSL 示例）

@Bean
public IntegrationFlow gcpFileProcessFlow() {
    return IntegrationFlow.from("notificationReceiveChannel")
            // 调用流式读取方法，返回Stream<S>
            .handle("fileReaderService", "readDataStream")
            // 拆分流为单个元素消息
            .split()
            // 按固定大小聚合成批次
            .aggregate(aggregatorSpec -> aggregatorSpec
                    .releaseStrategy(new MessageCountReleaseStrategy(100))
                    .expireGroupsUponCompletion(true)
                    // 最后不足100条的剩余数据，1秒无新消息自动释放
                    .groupTimeout(1000L)
            )
            // 批次消息直接发给路由
            .route(header(AppConstants.EVENT_HEADER_KEY))
            .get();
}

方案优势 #

和你当前手动实现的分批逻辑相比，框架原生方案无需自己维护计数、缓存、通道发送逻辑，天然支持错误重试、事务控制、监控统计能力，稳定性和可扩展性更强，同时全程内存中仅保留当前处理批次的少量数据，完全满足大文件读取不加载全量的要求。

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