JDK 26 Stream Gatherers 怎样组合流处理步骤

我第一次把 windowSliding、累积计算和结果映射放进同一条 Stream 管道时，最容易犯的错是把 Gatherer 当成另一个终止操作。实际上，JDK 26 里的 Stream.gather 是有状态的中间操作：它接收输入流，产出新的流，后面仍然可以继续 map、filter 或执行终止操作。

官方地址：https://docs.oracle.com/en/java/javase/26/docs/api/java.base/java/util/stream/Gatherers.html

组合 Stream Gatherers 的关键，是先确定每个 Gatherer 的输入和输出，再用 andThen 连接类型兼容的阶段；如果阶段依赖输入顺序，就不要为了“看起来更快”强行并行。

先确认 JDK 26 里的 API 位置

Oracle Java SE 26 API 文档把 Stream.gather、Gatherer 和 Gatherers 标为 Since 24，所以 JDK 26 项目可以直接使用，Java 24 及以上也可以使用同一组正式 API。它们仍属于 java.util.stream，不是另起一套流框架。

我会先把职责拆成三层：输入流负责提供元素，Gatherer负责有状态的中间变换，终止操作负责真正消费结果。这样排查泛型或生命周期问题时，不会把“组合失败”和“结果尚未消费”混在一起。

Stream.gather 为什么可以继续接管道

gather 返回的是新的 Stream，因此它的位置更接近 map 或 flatMap，而不是 collect。下面的例子先把数字分成固定窗口，再把每个窗口映射成一个摘要对象。

import java.util.List;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

record WindowSummary(int first, int last, int size) {}

class GatherDemo {
    static List summarize() {
        return Stream.of(10, 20, 30, 40, 50, 60, 70)
                // 每三个元素形成一个窗口，最后一个窗口可以不足三个
                .gather(Gatherers.windowFixed(3))
                // 继续使用普通 Stream 中间操作加工窗口
                .map(window -> new WindowSummary(
                        window.getFirst(),
                        window.getLast(),
                        window.size()))
                // 在这里才真正消费管道
                .toList();
    }
}

这里的输出类型从 Stream 变成了 Stream>，随后又变成 Stream。窗口本身是不可修改的 List，且窗口大小小于 1 会抛出 IllegalArgumentException。对业务来说，窗口大小就是一个需要明确记录的约束，而不是随手填的常量。

用 andThen 把两个 Gatherer 接起来

当两个 Gatherer 的类型可以衔接时，可以使用 andThen 得到一个组合 Gatherer。前一个 Gatherer 的输出会成为后一个 Gatherer 的输入。对我来说，最稳妥的写法是先分别写出两个阶段，再把组合表达式放进 gather，这样泛型报错更容易定位。

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

class ComposedGathererDemo {
    static List collectWindows() {
        Gatherer> windows =
                Gatherers.windowFixed(3);

        Gatherer, ?, String> format =
                Gatherers.map(window -> {
                    // 只把当前窗口格式化，避免改变窗口的生命周期
                    return "window=" + window;
                });

        return Stream.of(1, 2, 3, 4, 5, 6, 7)
                // 前一阶段的 List 输出必须匹配后一阶段的输入
                .gather(windows.andThen(format))
                .toList();
    }
}

这个例子里，组合后的类型是从 Integer 到 String。要特别注意，Gatherers.map 产生的是 Gatherer，不是普通 Stream.map 的别名；它适合把一个 Gatherer 输出继续交给另一个 Gatherer。

scan、window 和 mapConcurrent 怎么按负载选择

内置 Gatherers 可以按业务负载分成三组。窗口类适合批量、滑动区间和局部聚合；scan 适合把每次累计值都输出；mapConcurrent 适合每个元素都能独立处理、且需要限制并发度的映射。

如果后面还要组合，就优先让阶段的输入输出类型清楚；如果只是最后收集一个值，fold 可能比自定义状态更直观。对于外部 I/O，mapConcurrent 的并发度应按下游限流、连接数和超时一起决定，而不是只看 CPU 核数。

自定义 Gatherer 的四个生命周期部件

需要自定义逻辑时，Gatherer 的核心不是“写一个回调”这么简单。官方模型由 initializer、integrator、combiner 和 finisher 四个部件组成：创建状态、接收输入、合并分区状态、在输入结束时做最后动作。

import java.util.stream.Gatherer;

static Gatherer runningTotal() {
    class State {
        int total;
    }

    return Gatherer.ofSequential(
            // 每条顺序管道都从独立状态开始
            State::new,
            Gatherer.Integrator.ofGreedy((state, value, downstream) -> {
                // 累加后立即向下游推送当前前缀结果
                state.total += value;
                return downstream.push(state.total);
            })
    );
}

ofSequential 表示这个自定义 Gatherer 只按顺序处理。若要支持并行，需要提供语义正确的 combiner，并确认分区合并不会破坏业务顺序。不能因为状态对象是可变的，就直接把一个共享对象放到多个线程里。

并行、短路和资源边界怎么判断

Gatherer 能否并行，关键看是否提供了可用的合并函数。官方 API 允许对输入分区，分别处理后再合并中间状态；没有可靠合并语义的顺序操作应使用顺序 Gatherer。andThen 的组合也不能掩盖前一阶段对顺序的要求。

另外要区分两类结束：普通终止操作结束整个 Stream，Gatherer 的 finisher 则是在上游输入耗尽时做最后输出。若下游短路，不要假设所有输入都一定被读取；mapConcurrent 在下游不再需要结果时会尽力取消进行中的任务，外部资源仍需由调用方设计超时和关闭策略。

落地清单：组合前先看六件事

1. 确认运行环境至少是 Java 24，并在项目构建配置里明确版本。
2. 写出每个 Gatherer 的输入类型和输出类型，再使用 andThen。
3. 窗口大小、滑动步幅和最后不足一窗的语义要写进业务说明。
4. 需要逐项输出就考虑 scan，只要一个顺序结果再考虑 fold。
5. 自定义状态如果没有可证明的 combiner，优先使用 ofSequential。
6. 涉及外部调用时，给 mapConcurrent 设置受控并发度，并设计异常、超时和取消策略。

如果你的问题只是把普通的单元素转换串起来，Stream.map 仍然更简单；Gatherer 更适合窗口、状态、结束动作和受控并发这些普通中间操作表达不顺的场景。

相关问题

Gatherer 是不是 collect 的替代品？

不是。Gatherer 是中间操作，可以继续产生流并参与后续组合；collect 是终止操作，负责把流消费成结果。

windowFixed 产生的窗口能修改吗？

不能把它当作可变列表使用。官方文档说明生成的窗口是不可修改 List，需要修改时应显式复制。

为什么自定义 Gatherer 不应该直接共享状态？

因为并行评估可能为不同分区创建独立状态，再通过 combiner 合并。共享可变状态会破坏隔离，也让顺序和线程安全变得不可控。