Java Stream gatherers 怎么写自定义聚合：窗口收集、短路与结果类型

订单事件流里经常有一类需求：每收到三条事件就形成一个小批次，遇到“支付成功”后停止继续找，或者把一串输入折叠成一个带状态的结果。以前通常要在 collect、外部可变变量和手写循环之间做取舍。Java 24 的 Stream Gatherers 把这类“有状态的中间处理”放回了 Stream 管道里，但它不是 Collector 的替代品，窗口、短路和最终聚合要分开判断。

先把 gather 和 collect 的分工摆清楚

一次订单导入会产生连续事件：CREATED、PAID、ALLOCATED。如果目标是把所有事件最终装进一个不可变列表，collect 仍然直观；如果目标是“每三条输出一个窗口”或“命中某个事件后不再消费后续输入”，这就是中间阶段的状态转换，更适合 gather。

Oracle 的 Java 24 API 将 Stream.gather(Gatherer) 定义为有状态中间操作。Gatherer 可以把输入变成一对一、一对多或多对一的输出，也可以在处理过程中短路。这里先记住一个判断：结果要等全部输入结束后才产生，通常看 Collector 或 Gatherers.fold；结果要在管道中途持续产生，才看 gather。

List> batches = events.stream()
        .gather(Gatherers.windowFixed(3))
        .toList();

固定窗口：先用内置配方解决大多数批处理

windowFixed(3) 会按遇到顺序产生固定大小的列表。输入 8 条事件时，结果是三组完整窗口和一组只有两条的尾窗口。最后一组不会因为数量不足而被静默丢掉，这一点在批量落库和消息分发里很重要。

List events = List.of(
        "CREATED", "PAID", "ALLOCATED",
        "PICKED", "PACKED", "SHIPPED",
        "DELIVERED", "ARCHIVED");

List> windows = events.stream()
        .gather(Gatherers.windowFixed(3))
        .toList();

// [[CREATED, PAID, ALLOCATED], [PICKED, PACKED, SHIPPED],
//  [DELIVERED, ARCHIVED]]

窗口返回的列表是不可修改的。若下游需要补充批次号或重新排序，先复制成自己的集合，不要直接调用 add。另外，窗口大小越大，内存里同时保留的输入越多；小批量实时处理不应照搬离线导入的窗口值。

滑动窗口和 scan：同一份输入可以有不同观察方式

固定窗口是“消费三条，产出一组”；滑动窗口则会保留前一组的一部分，再加入新事件。例如窗口大小为 3 时，输入 A B C D 会观察到 A B C、B C D。它适合连续计算最近几次事件，但窗口重叠意味着同一事件可能被处理多次，写入外部系统前要先确认幂等策略。

List> recent = events.stream()
        .gather(Gatherers.windowSliding(3))
        .toList();

scan 则把上一步的状态持续传给下一步，类似“逐项产生前缀结果”。它和 fold 的差别很实用：scan 会不断向 downstream 发出中间结果，fold 更像直到输入结束才交付一个最终聚合值。选错会让下游误以为每个中间状态都是最终结果。

自定义 Gatherer：让“命中即停”成为管道的一部分

假设事件流来自一个昂贵的读取器，我们只想找到第一个 PAID 事件。普通 filter 只能筛选，不能表达“找到后停止后续读取”；takeWhile 又要求停止条件出现在目标之前。此时可以用一个带状态的 Gatherer，在 integrator 中把目标交给 downstream，并返回停止信号。

static Gatherer firstPaid() {
    return Gatherer.ofSequential(
            () -> new boolean[] { false },
            (seen, event, downstream) -> {
                if (seen[0]) {
                    return false;
                }
                if (event.equals("PAID")) {
                    seen[0] = true;
                    downstream.push(event);
                    return false;
                }
                return true;
            },
            (seen, downstream) -> { });
}

Optional paid = events.stream()
        .gather(firstPaid())
        .findFirst();

这段代码的关键不是“把循环藏起来”，而是把状态生命周期交给 Gatherer。ofSequential 明确表示这个实现没有并行合并语义；一旦业务要在并行流上使用，就必须重新设计状态合并，不能只把方法名换成并行版本。

结果类型和并行边界：先证明能合并，再谈性能

Gatherer 的三个类型参数分别对应输入元素、中间状态、输出元素。中间状态可以是计数器、窗口缓冲区或者流程阶段标记，输出的元素就是下游收集器能拿到的内容。别把可变状态暴露到 Gatherer 外部，也不要在 finisher 逻辑里偷偷修改已经交付出去的对象，这类写法跑串行测试时完全正常，一旦复用流管道就会出现非常难排查的状态污染问题。

要支持并行流就必须实现合规的 combiner 逻辑。两个分片的中间状态能不能无损合并，核心要看业务逻辑对元素出现顺序有没有要求：固定窗口聚合可以通过约定规则拼接边界处的跨分片窗口，这类场景适合并行，但类似「第一个匹配到 PAID 状态的元素」这类需求本身就强依赖元素遍历顺序，完全不适合随便开并行。找不到清晰的状态合并规则的时候，直接用顺序流配合自定义 Gatherer，是成本最低也最不容易出错的选择。

常见问题

Java Stream Gatherers 从哪个版本开始可用？

Java 24 的标准 API 已提供 Gatherer、Gatherers 和 Stream.gather。此前它在 Java 22、23 中经历过预览阶段，项目如果还停留在旧 JDK，需要先核对运行时和编译器版本。

windowFixed 的尾窗口会被丢弃吗？

不会丢。只要输入流不为空，最后末尾元素数量凑不满窗口大小的分组也会正常输出，不过返回的是不可修改的列表，要改内容的话自己复制一份新的再操作就行。

自定义 Gatherer 能直接用于 parallelStream 吗？

只有你自己写好了符合语义的 combiner，并且能验证分片合并之后业务规则依然完全成立的情况下，才建议用并行流跑自定义聚合。包含短路逻辑、对元素顺序敏感或者依赖外部副作用的 Gatherer，优先用顺序流处理就好。

小结

Stream Gatherers 的价值在于把“带状态的中间变换”表达成可组合的 Stream 步骤。简单分组先用 windowFixed，重叠观察用 windowSliding，持续输出状态看 scan，直到结尾才要一个聚合结果再看 fold。自定义 Gatherer 则要守住状态、短路和 combiner 三条边界，别为了新 API 把原本清楚的循环改复杂。