Java 中 Stream 流怎么在进行对象分批处理时利用自定义 Collector 实现 Batch 拆分-java教程-PHP中文网

java stream分批处理需自定义collector，核心是重写supplier、accumulator、combiner和finisher四函数，动态维护当前批次并按大小切片；示例实现支持线程安全与并行流，但推荐用streamex等成熟库避免错误。

java 中 stream 流怎么在进行对象分批处理时利用自定义 collector 实现 batch 拆分

Java Stream 中做对象分批处理时，用自定义 Collector 实现 Batch 拆分，核心是让收集器按固定大小把元素聚合成子列表，而不是等流结束才输出结果——这需要重写 supplier、accumulator、combiner 和 finisher 四个函数，并在累积过程中动态判断是否该“切片”。

理解 Collector 的四个组件如何协作完成分批

标准 Collector 是为单次归约设计的，但分批本质是“边累积、边截断、边产出”。关键在于：

supplier：返回一个可变容器（如 ArrayList<list>></list>），用于存所有批次；同时内部维护当前批次（ArrayList<t></t>）
accumulator：每次来一个元素，加到当前批次；若当前批次满（size == batchSize），就将它加入总批次列表，并新建空批次
combiner：合并两个 Collector 状态时，需把右侧的“未满批次”逐个添加到左侧对应位置（不能简单 addAll，否则会打乱批次边界）
finisher：最后检查当前批次是否非空，若存在就追加进去，再返回总批次列表

写一个通用的 batchCollector 工具方法

下面是一个线程安全、支持并行流的实现（注意：并行流下 combiner 必须正确合并中间状态）：

Gradle 9.8.0

下载

public static <t> Collector<t list>>> batchCollector(int batchSize) {
    return Collector.of(
        () -> {
            List<list>> batches = new ArrayList();
            List<t> currentBatch = new ArrayList();
            return new Object() {
                final List<list>> 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;
        }
    );
}</t></list></t></list></t></t>

使用示例：

List<string> data = Arrays.asList("a", "b", "c", "d", "e", "f", "g");
List<list>> batches = data.stream()
    .collect(batchCollector(3));
// 结果：[[a,b,c], [d,e,f], [g]]</list></string>