Java Gatherer怎样在 Stream 中维护窗口状态

我第一次把“每三个元素处理一次”塞进 Stream 时，先写了一个外部 List，再在计数器到 3 时手动清空。能跑，但状态散落在 lambda 外，遇到末尾不足一批或换成并行流就很容易误判。Java 的 Gatherer 把这类有状态的中间处理单独表达出来：窗口状态归 Gatherer 管，Stream 只负责把结果继续交给下游。

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

如果需求只是固定分批，先用 Gatherers.windowFixed(size)；如果每次要带上前一窗口的部分元素，改用 windowSliding(size)。只有在窗口规则需要自定义时，才值得自己实现 Gatherer。

先把窗口问题说清楚：固定窗口和滚动窗口

Stream.gather 是中间操作，可以把一个输入元素流变成另一个输出元素流。固定窗口按遇到顺序分组，窗口大小为 3 时，8 个元素会得到三个结果：[1, 2, 3]、[4, 5, 6]、[7, 8]。最后一个窗口不够 3 个并不是异常，而是 API 设计的一部分。

最小写法如下。代码注释保留了窗口大小、收集结果和示例输出的意图：

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

public class FixedWindowDemo {
    public static void main(String[] args) {
        // 每 3 个元素形成一个窗口；最后一组可以不足 3 个。
        List> windows = Stream.of(1, 2, 3, 4, 5, 6, 7, 8)
                .gather(Gatherers.windowFixed(3))
                .toList();

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

若改成 Gatherers.windowSliding(3)，结果会变为 [1,2,3]、[2,3,4]，直到 [6,7,8]。固定窗口适合批量提交、分片处理；滚动窗口适合计算最近 N 个样本。

自定义 Gatherer 的状态到底放在哪里

我觉得 Gatherer 最值得记住的不是 API 名字，而是状态边界：initializer() 创建一份私有状态，integrator() 每接收一个元素就更新状态并决定是否向下游输出，finisher() 在上游结束时处理还留在缓冲区里的元素。这样“窗口还差几个”的状态不会泄漏到外部变量。

下面是一个只保留完整窗口、忽略末尾残留的顺序 Gatherer 示例。它用于说明职责，不是替代内置窗口函数的通用实现：

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

static  Gatherer, List> fullWindows(int size) {
    // 顺序窗口不提供 combiner，避免把半窗口错误拼到一起。
    return Gatherer.ofSequential(
            ArrayList::new,
            (state, item, downstream) -> {
                // 状态只保存当前窗口，满 size 后复制一份再清空。
                state.add(item);
                if (state.size() == size) {
                    downstream.push(List.copyOf(state));
                    state.clear();
                }
                return true; // 继续接收上游元素。
            });
}

生产代码中仍要先检查 size > 0，并决定末尾不足窗口是保留、补齐还是丢弃。内置 windowFixed 会保留末尾窗口，而且产出的 List 不保证可修改；不要在下游直接调用 add 或 clear。

最容易踩的三个窗口边界

第一，窗口大小小于 1 会抛出 IllegalArgumentException，不要把 0 当作“关闭窗口”。第二，窗口是有状态中间操作，输入流没有元素时不会凭空产生一个空窗口。第三，并行不是默认收益：自定义 Gatherer 只有提供正确的 combiner，才有并行合并的语义；窗口依赖遇到顺序时，先保持顺序流更稳妥。

int size = 3;
if (size 

我的选择：先用内置窗口，再考虑自定义

批量发送、分段解析、固定大小写入通常直接使用 windowFixed；需要“最近 N 条”的重叠窗口使用 windowSliding。只有窗口完成条件不是单纯数量，例如遇到分隔符、超时或业务边界，才进入自定义 Gatherer，并把状态、输出时机和结束处理分别写清楚。

对我来说，Gatherer 的价值是把“状态如何流动”变成了可复用的中间步骤，而不是让 Stream 变得更复杂。先把窗口语义写成几个输入输出例子，再决定固定、滚动还是自定义，排错成本会低很多。

相关问题

windowFixed 和 windowSliding 有什么区别？
前者窗口之间不重叠，后者每个新窗口会带上前一个窗口的大部分元素。

最后一个不完整窗口会怎样？
内置固定窗口会输出它；自定义实现则由你的 finisher 或 integrator 逻辑决定保留还是丢弃。