Java Stream Gatherers 怎样实现自定义流聚合

如果一个 Stream 需要把元素按自定义规则聚成新的中间结果，Java 24 以后可以使用 Stream.gather(Gatherer)。它比单纯的 collect 更适合放在 Stream 中间：聚合器可以保存状态、持续向下游推送多个结果，也可以在输入结束时补发最后一个结果。

下面用“每 3 个元素组成一批”的场景实现一个自定义 Gatherer。核心做法是让 initializer 创建缓冲区，让 integrator 接收元素并在缓冲区装满时调用 Downstream.push，再用 finisher 处理不足一批的尾部数据。这个例子是有序聚合，因此先选择 ofSequential，不把没有定义清楚的合并语义硬塞进并行流。

官方资料：https://docs.oracle.com/en/java/javase/24/docs/api/java.base/java/util/stream/Gatherer.html

官方资料：https://docs.oracle.com/en/java/javase/24/docs/api/java.base/java/util/stream/Gatherers.html

先区分 collect 和 gather

collect 通常把整个 Stream 汇总成一个最终容器，例如一个 List 或 Map。固定批次场景则希望在流处理中间不断产出 List，后面的操作仍然可以继续使用这些批次。此时结果类型从 Stream 变成 Stream>，更接近一个可组合的中间操作。

Gatherer 的类型参数可以先这样记：T 是输入元素，A 是聚合过程中的状态，R 是推给下游的结果。它不是“自动把任意 lambda 变成并行安全代码”的快捷方式，状态如何创建、合并和收尾都要由实现者交代。

Gatherer 的四个协作点分别做什么

一个完整的 Gatherer 可以由四个部分组成。initializer 负责创建一份新的中间状态；integrator 每次接收一个输入元素，可以修改状态，也可以把结果推给下游；combiner 在允许并行化时合并两个局部状态；finisher 在上游结束后处理剩余状态。

对于固定批次，状态就是一个临时列表。列表达到批次大小后，先复制出一个快照再清空缓冲区。复制很重要：如果把仍会继续修改的缓冲区直接推给下游，后续追加会改变下游看到的内容。

用 ofSequential 实现固定批次

固定批次依赖输入的 encounter order，而且一个批次可能跨越并行分区边界。为了让示例的语义明确，使用 Gatherer.ofSequential，它只需要初始化器、整合器和收尾器，不声明一个并不存在的状态合并规则。

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Gatherer;
import java.util.stream.Stream;

public final class BatchGatherers {
    private BatchGatherers() {
        // 工具类不需要创建实例。
    }

    public static  Gatherer, List> fixedBatch(int size) {
        if (size  new ArrayList(size),
            (buffer, element, downstream) -> {
                // 每个输入元素只进入当前聚合状态一次。
                buffer.add(element);
                if (buffer.size()  batch = List.copyOf(buffer);
                buffer.clear();
                return downstream.push(batch);
            },
            (buffer, downstream) -> {
                // 上游结束时，把不足一整批的元素作为最后一批输出。
                if (!buffer.isEmpty()) {
                    downstream.push(List.copyOf(buffer));
                }
            }
        );
    }

    public static void main(String[] args) {
        List> batches = Stream.of(1, 2, 3, 4, 5, 6, 7, 8)
            .gather(fixedBatch(3))
            .toList();

        System.out.println(batches);
        // 输出示意：[[1, 2, 3], [4, 5, 6], [7, 8]]
    }
}

这里的整合器返回 boolean，返回值来自 downstream.push。当下游不再需要数据时，返回 false 可以让 Gatherer 停止继续接收输入；不要为了“看起来简单”而无视这个信号。

为什么 finisher 不能省略

输入长度是 8、批次大小是 3 时，完整批次只有两组，最后的 7、8 会留在状态列表中。如果没有 finisher，这两个元素不会自然变成输出，结果就会悄悄丢数据。

收尾器只做一件事：检查缓冲区是否为空，非空就复制成一个独立列表并推给下游。使用 List.copyOf 还可以让已推送的批次不再被内部缓冲区的后续清空或追加影响。输出列表是不可修改视图，后续逻辑如果需要修改，应显式复制。

先用内置 Gatherers，再决定是否自定义

如果需求只是常见的窗口或累计，不必重复实现。Java 24 的 Gatherers 已提供几种中间操作：

自定义 Gatherer 的价值在于“输出时机或状态结构”不同，例如要按业务边界切批、遇到特殊元素提前结束、对每批执行状态判断，或者需要把两个已有 Gatherer 用 andThen 组合起来。

并行化前先把 combiner 语义说清楚

Gatherer.of 会接收 combiner，表示多个局部状态可以合并；ofSequential 则明确告诉库只按顺序评估。固定批次并不是把两个列表简单拼接就结束，因为分区左侧的半批数据可能要和右侧分区的开头元素合成一批，且输出顺序必须保持一致。

如果不能清楚定义“局部状态如何合并、边界批次如何跨分区传递、收尾时哪些结果仍有效”，就继续使用顺序 Gatherer。等到确实需要并行，再为状态设计完整的合并协议，并用下游短路、空输入、恰好整除和尾部不足等场景分别验证。

还要注意状态对象的生命周期：不要把 Gatherer 收到的状态或 Downstream 引用保存到方法调用之外，更不要把可变缓冲区暴露给其他线程。Gatherer 的状态是聚合实现的内部细节，不是跨请求共享缓存。

常见问题

Java 版本低于 24 能直接使用 Gatherer 吗？

本文使用的是 Java SE 24 API 中的 Gatherer、Gatherers 和 Stream.gather。如果项目运行时或编译环境低于对应版本，应先确认目标 JDK 是否提供这些 API，不要只升级 IDE 的语法提示。

固定批次是否应该直接用 collect？

如果只需要一个最终的二维列表，collect 也可以完成；如果批次要继续进入后续 Stream 操作，或要在满批时持续向下游产出，gather 的中间操作语义更贴合需求。

为什么不在 integrator 中直接复用 buffer 推送？

因为 buffer 在下一次整合时还会被清空和追加。推送前复制出批次快照，才能让已经产生的结果拥有稳定内容。

什么时候可以改用 Gatherers.windowFixed？

只要需求就是按固定大小产生有序窗口，且接受内置实现的不可修改列表语义，优先使用 Gatherers.windowFixed。只有当批次边界、输出条件或状态更新明显不同，才值得维护自定义 Gatherer。