Spring Batch如何批量处理Chunk中的Item并调用外部API

你当前的Spring Batch Job已经按10个Item为一个Chunk分割数据，但默认的Processor会逐个处理Item。要实现将整个Chunk的Item聚合成列表后批量调用外部API，最直接且符合框架设计的方式是在ItemWriter中处理批量请求，因为ItemWriter的write方法本身就接收整个Chunk的Item列表。

方案一：直接在ItemWriter中实现批量调用 #

Spring Batch的ItemWriter接口定义的write方法参数是List<? extends T>，刚好对应一个Chunk的所有数据。你只需要重写ItemWriter的逻辑，直接将整个列表传给外部API即可，无需修改原有的Chunk配置。

修改后的代码示例：

首先定义批量处理的ItemWriter：

@Bean
public ItemWriter<Item> itemWriter(ExternalApiService externalApiService) {
    return items -> {
        // 直接使用整个Chunk的Item列表调用外部API
        externalApiService.batchProcess(items);
    };
}

如果不需要逐个处理Item，可以直接去掉Processor，简化Job配置：

@Bean
public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) {
    return jobs.get("job")
            .start(steps.get("step")
                    .<Item, Item>chunk(10)
                    .reader(itemReader())
                    // 若无需单个Item预处理，可移除processor
                    // .processor(itemProcessor())
                    .writer(itemWriter())
                    .build())
            .build();
}

方案二：在Processor阶段聚合（适用于需全Chunk预处理场景） #

如果必须在Processor阶段对整个Chunk的数据做预处理后再调用API，可以自定义一个带状态的Processor，结合@AfterChunk注解在Chunk处理完成后执行批量操作。注意需要给Processor添加@StepScope保证每个Step实例的状态隔离：

自定义聚合Processor：

@StepScope
@Component
public class AggregatingItemProcessor implements ItemProcessor<Item, List<Item>> {

    private final List<Item> chunkItems = new ArrayList<>();
    private final ExternalApiService externalApiService;

    // 构造注入外部API服务
    public AggregatingItemProcessor(ExternalApiService externalApiService) {
        this.externalApiService = externalApiService;
    }

    @Override
    public List<Item> process(Item item) {
        // 收集当前Chunk的所有Item
        chunkItems.add(item);
        // 返回null，暂不输出结果
        return null;
    }

    @AfterChunk
    public void afterChunkProcessing(ChunkContext chunkContext) {
        // Chunk处理完成后，批量调用外部API
        externalApiService.batchProcess(chunkItems);
        // 清空列表，准备下一个Chunk
        chunkItems.clear();
    }
}

修改Job配置：

调整Step的泛型为<Item, List<Item>>，Writer可留空或做后续收尾处理：

@Bean
public Job job(JobBuilderFactory jobs, StepBuilderFactory steps, AggregatingItemProcessor aggregatingItemProcessor) {
    return jobs.get("job")
            .start(steps.get("step")
                    .<Item, List<Item>>chunk(10)
                    .reader(itemReader())
                    .processor(aggregatingItemProcessor)
                    .writer(items -> { /* 无需额外处理，批量操作已在AfterChunk中完成 */ })
                    .build())
            .build();
}

方案对比 #

• 方案一更简洁，符合Spring Batch的设计初衷（Writer负责批量数据输出），无需管理状态，推荐优先使用。
• 方案二适用于必须对整个Chunk数据做预处理的场景，但需要处理状态隔离，代码相对复杂。

内容的提问来源于stack exchange，提问作者The Anh Nguyen