Java Stream Gatherer 怎么实现有状态中间操作

需要让 Stream 中间操作记住上一条元素、累计当前分组，或者在输入结束时补发一个结果时，普通的 map 和 filter 就不够了。Java Stream Gatherer 可以把这段状态逻辑封装进 Stream 链：initializer 创建私有状态，integrator 逐个接收输入，finisher 处理流结束时还没输出的尾部状态。

官方文档：https://docs.oracle.com/en/java/javase/26/docs/api/java.base/java/util/stream/Gatherer.html

先把“状态”和“输出时机”分开

下面用连续事件分组说明。输入是 A, A, B, B, C，输出应为 A:[1,2]、B:[3,4]、C:[5]。这里的“当前分组”是中间状态，而不是最终结果；只有看到下一个 key，才能知道上一组已经结束。

用 ofSequential 写一个有状态中间操作

ofSequential 适合这种必须保持遇到顺序的操作。每次 key 改变时先把旧状态复制成不可变列表并推给下游，再清空列表开始新组；输入结束时由 finisher 补上最后一组。

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

record Event(String key, int value) {}
record Group(String key, List values) {}

final class State {
    String key;
    final List values = new ArrayList();
}

Gatherer consecutiveGroups = Gatherer.ofSequential(
    // 为一次 gathering 创建私有可变状态，不从外部捕获共享列表
    State::new,
    (State state, Event event, Gatherer.Downstream downstream) -> {
        // key 改变说明上一段连续分组已经结束
        if (state.key != null && !state.key.equals(event.key())) {
            downstream.push(new Group(state.key, List.copyOf(state.values)));
            state.values.clear();
        }
        state.key = event.key();
        state.values.add(event.value());
        return true; // 返回 false 才会请求短路；本例要读完整个流
    },
    (State state, Gatherer.Downstream downstream) -> {
        // 没有下一个 key 可触发输出，所以在流结束时刷出尾组
        if (state.key != null) {
            downstream.push(new Group(state.key, List.copyOf(state.values)));
        }
    }
);

List result = Stream.of(
        new Event("A", 1), new Event("A", 2),
        new Event("B", 3), new Event("B", 4), new Event("C", 5))
    .gather(consecutiveGroups)
    .toList();
// result：[{key=A, values=[1, 2]}, {key=B, values=[3, 4]}, {key=C, values=[5]}]

这里的返回值不是“处理成功”的标志，而是是否继续接收输入的信号。连续分组永远返回 true，因为它要读完整个流；需要找到第一个满足条件的元素时，才可以在 push 后返回 false 实现短路。

为什么这里不直接改成 parallel

连续分组的结果取决于相邻元素的 encounter order。把输入切成两段后，第一段末尾的 A 可能要和第二段开头的 A 合并；这要求 combiner 不仅合并列表，还要知道两段的首尾 key，并处理跨分区合并。若没有清晰、可证明的合并规则，强行并行只会把顺序问题藏起来。

因此本例使用 Gatherer.ofSequential，它明确表达“状态操作按顺序消费”。如果业务是前缀扫描、固定窗口等常见模式，可以优先查看 Gatherers.scan、Gatherers.windowFixed 和 Gatherers.windowSliding；如果确实需要并行，则应使用完整的 initializer、integrator、combiner、finisher 组合，并单独验证跨分区语义。

常见问题

状态能不能定义成方法外的静态列表？

不建议。外部可变状态会让多个 Stream 运行互相污染，也破坏 Gatherer 对状态生命周期的管理；把状态放进 initializer 创建的对象更安全。

为什么最后一组必须写在 finisher？

最后一组后面没有“下一个 key”触发切换，integrator 不会自动知道输入已经结束，所以必须在 finisher 中显式 push。

什么时候可以自定义 combiner？

只有当两个分区状态能够按明确规则合并，并且合并后的结果与顺序语义一致时才适合自定义；连续分组通常先采用顺序实现。

判断一个有状态中间操作是否适合 Gatherer，可以先问三个问题：状态是否只属于当前 gathering、输出是否可能延迟到流结束、分区状态是否有可靠的合并规则。前两个问题回答“是”并不意味着必须并行，顺序边界本身也是设计结果。