Spring Batch 中实现批量项分组处理以减少外部 API 调用次数-java教程-PHP中文网

Spring Batch 中实现批量项分组处理以减少外部 API 调用次数

本文介绍如何在 Spring Batch 中将原始数据按指定大小（如每 10 条）分组为 List，使 Processor 和 Writer 能以“整块”为单位处理，从而避免单条记录发起一次 API 请求，显著提升性能与资源利用率。

本文介绍如何在 spring batch 中将原始数据按指定大小（如每 10 条）分组为 list，使 processor 和 writer 能以“整块”为单位处理，从而避免单条记录发起一次 api 请求，显著提升性能与资源利用率。

在标准 Spring Batch 流程中，chunk(10) 仅控制写入批次大小，而 ItemProcessor 仍默认对每个 Item 单独调用——这导致即使配置了 10 条一批，仍会触发 10 次独立的处理器逻辑和外部 API 调用。要真正实现“以批为单位处理”，关键在于让整个 chunk 成为一个处理单元，即：让 ItemReader 直接返回 List<item></item>，而非单个 Item。

为此，需重构数据流类型签名，并自定义 ItemReader<list>></list>：

✅ 正确配置：统一类型链路

@Bean
public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) {
    return jobs.get("batchJob")
            .start(steps.get("processingStep")
                    .<list>, List<item>>chunk(1) // ⚠️ chunk size = 1，因 reader 已输出「完整 chunk」
                    .reader(itemReader())
                    .processor(itemProcessor())
                    .writer(itemWriter())
                    .build())
            .build();
}</item></list>

注意泛型 <list>, List<item>></item></list>：表示 reader 输出 List<item></item>，processor 输入/输出也均为 List<item></item>，确保类型全程一致。

✅ 自定义 ChunkItemReader：按需切片数据

@Bean
@StepScope
public ItemReader<list>> itemReader(@Value("#{jobParameters['items']}") List<item> items) {
    return new ChunkItemReader(items, 10); // 每次 read() 返回 10 条组成的 List
}</item></list>

ChunkItemReader 实现如下（线程安全、无状态、支持重启）：

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 = Collections.unmodifiableList(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 new ArrayList(chunk); // 返回副本，避免 subList 引用问题
    }
}</t></t></t></t></list></t>

? 关键点：subList 返回的是原列表的视图，若后续修改原列表可能引发 ConcurrentModificationException。因此推荐使用 new ArrayList(chunk) 创建独立副本。

✅ Processor 与 Writer 的适配

此时 ItemProcessor 类型应为：

@Bean
public ItemProcessor<list>, List<item>> itemProcessor() {
    return chunk -> {
        // ✅ chunk 是包含 10 个 Item 的 List
        List<item> processed = callExternalApiInBatch(chunk); // 单次批量 API 调用
        return processed;
    };
}</item></item></list>

同理，ItemWriter<list>></list> 可直接遍历处理或透传结果：

@Bean
public ItemWriter<list>> itemWriter() {
    return chunks -> chunks.forEach(chunk -> {
        // 可选：持久化批量结果、发送通知等
        System.out.println("Wrote batch of " + chunk.size() + " items");
    });
}</list>

⚠️ 注意事项

状态管理：ChunkItemReader 依赖 currentIndex 维护读取位置，必须确保其在 step 重启时能正确恢复（当前实现不支持自动重启）。如需容错，建议改用 JdbcPagingItemReader 或 JdbcCursorItemReader 配合 SQL 分页 + 自定义 RowMapper 合并成块。
内存考量：若原始数据量极大（如百万级），避免一次性加载全部 items 到内存；应结合数据库游标或分页查询，在 read() 中动态拉取下一批。
事务边界：chunk(1) 表示每个「chunk 列表」为一个事务单元，API 调用失败将回滚整个批次，符合一致性要求。