Java Stream分批处理需自定义Collector，核心是重写supplier、accumulator、combiner和finisher四函数，动态维护当前批次并按大小切片；示例实现支持线程安全与并行流，但推荐用StreamEx等成熟库避免错误。
Java Stream 中做对象分批处理时，用自定义 Collector 实现 Batch 拆分，核心是让收集器按固定大小把元素聚合成子列表，而不是等流结束才输出结果——这需要重写 supplier、accumulator、combiner 和 finisher 四个函数，并在累积过程中动态判断是否该“切片”。
理解 Collector 的四个组件如何协作完成分批
标准 Collector 是为单次归约设计的，但分批本质是“边累积、边截断、边产出”。关键在于：
-
supplier：返回一个可变容器（如
ArrayList<List<T>>），用于存所有批次；同时内部维护当前批次（ArrayList<T>） - accumulator：每次来一个元素，加到当前批次；若当前批次满（size == batchSize），就将它加入总批次列表，并新建空批次
- combiner：合并两个 Collector 状态时，需把右侧的“未满批次”逐个添加到左侧对应位置（不能简单 addAll，否则会打乱批次边界）
- finisher：最后检查当前批次是否非空，若存在就追加进去，再返回总批次列表
写一个通用的 batchCollector 工具方法
下面是一个线程安全、支持并行流的实现（注意：并行流下 combiner 必须正确合并中间状态）：
public static <T> Collector<T, ?, List<List<T>>> batchCollector(int batchSize) {
return Collector.of(
() -> {
List<List<T>> batches = new ArrayList<>();
List<T> currentBatch = new ArrayList<>();
return new Object() {
final List<List<T>> batches = batches;
final List<T> currentBatch = currentBatch;
};
},
(acc, item) -> {
acc.currentBatch.add(item);
if (acc.currentBatch.size() == batchSize) {
acc.batches.add(new ArrayList<>(acc.currentBatch));
acc.currentBatch.clear();
}
},
(acc1, acc2) -> {
// 把 acc2 的 currentBatch 元素逐个加到 acc1
acc2.currentBatch.forEach(acc1.currentBatch::add);
// 若 acc1 当前批次满了，触发切片
while (acc1.currentBatch.size() >= batchSize) {
acc1.batches.add(new ArrayList<>(acc1.currentBatch.subList(0, batchSize)));
acc1.currentBatch = new ArrayList<>(acc1.currentBatch.subList(batchSize, acc1.currentBatch.size()));
}
return acc1;
},
acc -> {
if (!acc.currentBatch.isEmpty()) {
acc.batches.add(new ArrayList<>(acc.currentBatch));
}
return acc.batches;
}
);
}
使用示例：
立即学习“Java免费学习笔记（深入）”；
List<String> data = Arrays.asList("a", "b", "c", "d", "e", "f", "g");
List<List<String>> batches = data.stream()
.collect(batchCollector(3));
// 结果：[[a,b,c], [d,e,f], [g]]
更简洁安全的替代方案：用第三方库或封装辅助类
手写 Collector 容易出错（尤其 combiner 逻辑）。实际项目中推荐：
- 用
guava的Lists.partition(list, size)—— 但它要求先转成 List，无法流式处理大集合 - 用
StreamEx库的StreamEx.of(stream).batch(batchSize)，原生支持流式分批 - 自己封装一个
BatchingSpliterator，配合StreamSupport.stream()使用，避免 Collector 复杂度
注意事项和常见陷阱
分批 Collector 在并行流中要格外小心：
- 不要在 accumulator 中直接操作共享的
List<List<T>>，必须用线程局部状态（如上面的匿名对象） - combiner 必须能处理任意两个中间状态的合并，不能假设“左边已切片、右边是新批次”
- batchSize ≤ 0 时应抛异常，避免无限循环或空指针
- 如果元素类型有 null 值，确保不因 null 导致
NullPointerException
