Spring Batch 中实现批量分块处理并聚合调用外部 API 的完整方案-java教程-PHP中文网

Spring Batch 中实现批量分块处理并聚合调用外部 API 的完整方案

本文介绍如何在 Spring Batch 中将原始数据按指定大小（如每 10 条）聚合成 List 批量处理，避免逐条调用外部 API，从而显著提升性能和资源利用率。核心思路是自定义 ItemReader> 实现分块读取，配合 chunk(1) 驱动整块数据流式处理。

本文介绍如何在 spring batch 中将原始数据按指定大小（如每 10 条）聚合成 list 批量处理，避免逐条调用外部 api，从而显著提升性能和资源利用率。核心思路是自定义 `itemreader>` 实现分块读取，配合 `chunk(1)` 驱动整块数据流式处理。

在 Spring Batch 默认的 chunk 模式中，即使配置 .chunk(10)，ItemProcessor 仍以单个元素（Item）为单位被调用 10 次，这导致后续若需批量调用外部 API（如 REST 接口支持批量提交），会引发 10 次 HTTP 请求，造成严重性能浪费与服务压力。真正的批量处理应让 ItemProcessor 接收一个 List<item></item>，一次性完成聚合逻辑与外部调用。

解决方案的关键在于反转职责：不再依赖框架对单个 item 的自动分块，而是由 ItemReader 主动返回预组装的“数据块”（即 List<item></item>），再将 chunk size 设为 1 —— 此时每个 chunk 就是一组批量数据。

✅ 正确配置示例

@Bean
public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) {
    return jobs.get("batchProcessingJob")
            .start(steps.get("processChunkedItems")
                    .<list>, List<item>>chunk(1) // 注意：输入/输出类型均为 List<item>
                    .reader(itemReader())
                    .processor(itemProcessor())
                    .writer(itemWriter())
                    .build())
            .build();
}

@Bean
@StepScope
public ItemReader<list>> itemReader() {
    // 假设 items 已从数据库、文件或其它来源加载完成
    List<item> allItems = loadAllItems(); // 你的数据源加载逻辑
    int chunkSize = 10;
    return new ChunkItemReader(allItems, chunkSize);
}</item></list></item></item></list>

✅ 自定义分块读取器：`ChunkItemReader`

import org.springframework.batch.item.ItemReader;
import java.util.List;

public class ChunkItemReader<t> implements ItemReader<list>> {

    private final List<t> items;
    private final int chunkSize;
    private int currentIndex = 0;

    public ChunkItemReader(List<t> items, int chunkSize) {
        this.items = items;
        this.chunkSize = chunkSize;
    }

    @Override
    public List<t> read() {
        if (currentIndex >= items.size()) {
            return null; // 数据耗尽
        }
        int endIndex = Math.min(currentIndex + chunkSize, items.size());
        List<t> chunk = items.subList(currentIndex, endIndex);
        currentIndex = endIndex;
        return chunk;
    }
}</t></t></t></t></list></t>

⚠️ 注意：subList 返回的是原列表的视图（view），非独立副本。如需线程安全或防止后续修改影响原始数据，可在 read() 中显式复制：new ArrayList(items.subList(...))。

✅ Processor 与 Writer 的适配

此时 ItemProcessor 类型应为 ItemProcessor<list>, List<item>></item></list>，可直接对整批数据执行批量 API 调用：

@Bean
public ItemProcessor<list>, List<item>> itemProcessor() {
    return items -> {
        // ✅ 一次调用外部批量接口
        List<itemresult> results = externalApi.batchProcess(items);

        // 可选：将结果映射回原 items 或生成新对象
        return results.stream()
                .map(this::toProcessedItem)
                .collect(Collectors.toList());
    };
}</itemresult></item></list>

同理，ItemWriter<list>></list> 也接收 List<item></item>，适合做批量入库、日志记录等操作。

? 补充说明与最佳实践

内存友好性：该方案适用于全量数据可预加载至 JVM 内存的场景（如数万级 item）。若数据量极大（百万+），建议改用基于游标/分页的流式 ItemReader（如 JdbcCursorItemReader）配合自定义分块逻辑。
事务边界：每个 chunk（即每个 List<item></item>）仍处于同一事务中，确保原子性；但注意外部 API 调用本身不参与 Spring Batch 事务管理，需自行设计幂等性与重试策略。
扩展性提示：可将 ChunkItemReader 泛化为支持 Iterator<t></t> 或 Stream<t></t> 输入，进一步解耦数据源。