Java Stream gatherer 的状态与短路边界

我第一次把自定义 Stream 变换迁移到 gather 时，最容易混淆的是“状态保存在哪里”和“什么时候应该停止”。可以先记住一个结论：Gatherer 的 A 是中间状态，integrator 负责接收输入并按需产出，finisher 负责处理输入结束时仍留在状态里的尾部数据；下游返回拒绝时，整合器应当停止继续推送。

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

这个 API 从 Java 24 起进入正式 API。迁移时不要把它当成“带几个回调的 Collector”：它是 Stream 中间操作，既可以有状态，也可以短路，还可以在提供正确合并函数时参与并行处理。

先把 gatherer 看成一组协作点

一个 gatherer 可以拆成四个职责：initializer() 创建状态，integrator() 消化一个输入元素，combiner() 合并不同分区的状态，finisher() 在上游结束时把尾部结果推给下游。状态只应由这些回调在调用期间使用，不要保存到别的线程或跨调用暴露。

如果操作依赖输入顺序，最稳妥的起点是 Gatherer.ofSequential。例如把数字按固定大小分窗，窗口未满时只保存状态，窗口满或输入结束时才输出：

static Gatherer, List> fixedWindow(int size) {
    if (size  {
            state.add(value); // 状态只保存当前未完成窗口
            if (state.size()  window = List.copyOf(state);
            state.clear(); // 先清空，再把不可变副本交给下游
            return downstream.push(window); // 下游拒绝时把短路信号向上返回
        },
        (state, downstream) -> {
            if (!state.isEmpty()) {
                downstream.push(List.copyOf(state)); // 收尾输出不足一个窗口的尾部数据
            }
        });
}

这里的 ArrayList 是每次 gather 操作的局部状态，不是共享缓存。List.copyOf 让已经推送的窗口不再跟随后续的 clear() 变化；如果省略复制，读者看到的结果就可能与状态生命周期纠缠。

短路不是异常，而是下游拒绝继续接收

Downstream.push 返回 false 后，后续推送都应视为不再被接受；整合器返回 false，则表示没有更多元素需要传给它。两者都不是异常，也不应该用异常来实现“找到第一个结果后停止”。isRejecting() 是一个尽力而为的提示，可以减少无意义的工作，但真正的边界仍然是 push 的返回值。

无状态的逐元素转换可以使用并行化的工厂方法，代码重点是把下游的返回值原样交还给 Stream：

static  Gatherer map(Function super T, ? extends R> mapper) {
    return Gatherer.of((unused, element, downstream) -> {
        R mapped = mapper.apply(element); // 先完成当前元素的转换
        return downstream.push(mapped); // 让下游决定是否还需要后续元素
    });
}

业务回调本身抛出的异常仍然会按 Stream 管道的异常语义传播；“短路”只表示下游不再要数据，不代表把异常吞掉。对于有资源的回调，也要在回调自己的边界内完成关闭，不能把资源句柄塞进可合并状态后交给未知线程。

并行边界要看 combiner 是否成立

选择并行版本前，我会先问一个问题：两个输入分区各自得到的状态，能否在不改变结果顺序和语义的前提下合并？如果答案是否定的，就使用 ofSequential，不要为了“支持 parallelStream”硬写一个看似能运行的 combiner。

迁移检查可以压缩成五项：状态是否只在当前 gatherer 内使用；每次 push 是否检查返回值；输入结束时是否处理尾部状态；combiner 是否真正保持语义；并行时是否仍然满足顺序、隔离和资源生命周期约束。少掉其中一项，代码可能在小样本上正常，却在短路或分区时产生难以定位的差异。

常见追问

gatherer 的状态能不能被多个线程共享？

不应该主动共享。并行实现会为分区创建并隔离中间状态，再通过合并函数组合；回调也不应长期保存状态或下游对象。

finisher 一定会输出数据吗？

不一定。它只是输入结束时的收尾钩子；没有尾部数据时可以不推送，默认 finisher 也表示没有额外结束动作。

integrator 返回 false 后还要继续处理当前元素吗？

不需要。返回 false 表示该 gatherer 不再接受更多输入；如果已经调用了 push，还应尊重下游返回的 false，避免继续制造结果。