本文介绍如何在 Spring Batch 中将原始数据按指定大小（如每 10 条）聚合成 List 批量处理，使 ItemProcessor 接收整个 chunk 而非单个 item，从而减少外部 API 请求次数，显著提升批处理性能与资源利用率。
本文介绍如何在 spring batch 中将原始数据按指定大小（如每 10 条）聚合成 list 批量处理，使 `itemprocessor` 接收整个 chunk 而非单个 item，从而减少外部 api 请求次数，显著提升批处理性能与资源利用率。
在标准 Spring Batch 流程中，chunk(10) 表示「每次读取 10 条记录 → 逐条调用 ItemProcessor → 汇总后交由 ItemWriter 写入」。但若业务逻辑需将整批数据（如 10 个 item）一次性提交至外部 REST API，默认的逐项处理模式会导致 10 次 HTTP 请求，造成严重性能瓶颈和连接开销。
解决思路是：反转职责边界——让 ItemReader 直接返回已分片的 List<item></item>，而非单个 Item。此时 chunk 配置应设为 .chunk(1)，因为 Reader 每次输出一个“逻辑批次”，而 Processor 和 Writer 均以 List<item></item> 为单位进行操作。
✅ 正确配置示例
@Bean
public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) {
return jobs.get("batchJob")
.start(steps.get("processChunkStep")
.<List<Item>, List<Item>>chunk(1) // ⚠️ 注意：chunk size = 1，因 reader 已返回完整 chunk
.reader(itemReader())
.processor(itemProcessor())
.writer(itemWriter())
.build())
.build();
}
@Bean
@StepScope
public ItemReader<List<Item>> itemReader() {
// 假设 items 已从数据库/文件加载完成（可在 @StepScope Bean 中注入 Repository）
List<Item> allItems = loadAllItems(); // e.g., jdbcTemplate.query(...)
int chunkSize = 10;
return new ChunkItemReader<>(allItems, chunkSize);
}? 自定义分块读取器 ChunkItemReader
该类继承 ItemReader<list>></list>，内部维护索引并按需切片：
public class ChunkItemReader<T> implements ItemReader<List<T>> {
private final List<T> items;
private final int chunkSize;
private int currentIndex = 0;
public ChunkItemReader(List<T> items, int chunkSize) {
this.items = Objects.requireNonNull(items);
this.chunkSize = Math.max(1, 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; // 返回当前批次（不可变视图，如需安全复制可用 new ArrayList<>(chunk)）
}
}? 关键说明：
subList()返回的是原列表的视图，若后续 processor/writer 会修改内容且需线程安全或独立副本，请显式构造新ArrayList。
? Processor 与 Writer 的适配
此时 ItemProcessor 类型需声明为：
public class BatchApiProcessor implements ItemProcessor<List<Item>, List<Item>> {
@Override
public List<Item> process(List<Item> chunk) throws Exception {
// ✅ 一次性调用外部 API：POST /api/batch with chunk
apiClient.batchProcess(chunk);
return chunk; // 或返回增强后的结果（如添加响应字段）
}
}同理，ItemWriter 应接收 List<list>></list>（注意：Spring Batch 在 chunk 模式下会将多个 List<item></item> 聚合成 List<list>></list>），因此建议使用泛型更清晰的写法：
@Bean
public ItemWriter<List<Item>> itemWriter() {
return chunks -> {
for (List<Item> chunk : chunks) {
// 处理每个 chunk（例如记录日志、触发回调等）
log.info("Processed batch of {} items", chunk.size());
}
};
}⚠️ 注意事项与最佳实践
-
内存友好性：确保
loadAllItems()不加载超大数据集到内存；对于海量数据，应改用分页查询 +JdbcPagingItemReader并在 reader 层做流式分块。 -
事务边界：每个 chunk（即每个
List<item></item>）仍处于同一事务中，失败时整批回滚，符合幂等设计预期。 -
错误处理：若某批 API 调用失败，可通过
SkipPolicy或FaultTolerantStepBuilder实现跳过或重试，但需保证外部 API 支持粒度可控的失败反馈（如返回部分失败 item ID）。 -
类型一致性：务必统一泛型签名 ——
ItemReader<list>></list>,ItemProcessor<list>, List<item>></item></list>,ItemWriter<list>></list>，避免运行时ClassCastException。
通过该方案，你将以最小侵入方式突破 Spring Batch 默认的“单 item 流水线”限制，真正实现高效、可扩展的批量集成场景。
