Java Stream Gatherer 怎么把多步流处理收成可复用组件：状态边界、并行语义与验收

批量导入会员数据时，校验、分组和结果汇总经常被塞进一条很长的 Stream 链里。代码能跑，但一旦要同时保留失败原因、限制批次大小，或者把同一段处理复用到另一条流，临时变量和边界判断就会迅速变多。Java Stream Gatherer 提供了一个更明确的扩展点：把“流元素如何进入状态、什么时候向下游发结果、最后如何收尾”集中到一个可测试的组件里。

先看一个难维护的批量校验链

假设每 100 条会员记录组成一个小批次，批次满了就交给下游处理，格式错误的记录还要保留行号和原因。只用普通 map、filter 和 collect 当然可以完成，但状态通常藏在外部可变集合里：

List> batches = new ArrayList();
List current = new ArrayList();
for (Member member : members) {
    if (!isValid(member)) {
        continue;
    }
    current.add(member);
    if (current.size() == 100) {
        batches.add(List.copyOf(current));
        current.clear();
    }
}
if (!current.isEmpty()) {
    batches.add(List.copyOf(current));
}

这段逻辑的问题不在于长，而在于“收集状态”和“向下游发出结果”没有一个稳定的抽象。后续如果要在每批结果生成审计记录，或者达到错误阈值后停止读取，修改点会散落在循环里。

Gatherer 到底接管了哪些动作

可以把一个 Gatherer 想成一台小型流处理器：它拥有自己的状态容器，接收上游元素，并在合适的时机把结果交给下游。典型实现会提供四个部分：创建状态的 initializer、处理每个输入的 integrator、正常结束时的 finisher，以及描述并行组合方式的 combiner。

下面用“每三条有效记录形成一个批次”说明结构。示例中的 Batch、Member 和 BatchResult 是业务对象，重点在边界，不在于复制成品类定义。

static Gatherer, BatchResult> batchesOf(int size) {
    return Gatherer.ofSequential(
        ArrayList::new,
        (state, member, downstream) -> {
            if (!isValid(member)) {
                return true;
            }
            state.add(member);
            if (state.size() == size) {
                downstream.push(new BatchResult(List.copyOf(state)));
                state.clear();
            }
            return true;
        },
        (state, downstream) -> {
            if (!state.isEmpty()) {
                downstream.push(new BatchResult(List.copyOf(state)));
            }
        });
}

这里的 true 表示继续接收上游元素。若下游已经不需要更多结果，或者业务错误达到停止阈值，integrator 可以返回 false，让这条处理路径提前结束。finisher 则只负责处理“最后一个不满批次”，不能假设它一定会收到完整批次。

状态容器必须有清晰的所有权

state 只属于当前 Gatherer 实例。推送结果时用 List.copyOf 固化快照，再清空状态；如果直接把可变列表交给下游，下一轮收集会让已经输出的批次内容发生变化。这类问题通常不会在单条数据测试里暴露，却会在异步消费或日志序列化时变得很难定位。

为什么并行语义不能顺手打开

上面的实现明确使用了顺序 Gatherer，因为“每三条组成一个批次”依赖输入顺序。把它直接放进 parallel() 后，多个分片可能各自形成局部批次，最后没有安全的方式把边界重新拼回原始顺序。即使最终元素数量正确，批次成员也可能不同。

只有在状态可以独立拆分、结果允许重排，或者你已经定义了可靠的合并规则时，才应该设计 combiner。例如计数器、按键聚合这类状态通常更容易合并；带严格顺序的窗口、首尾依赖和有副作用的批处理则应先保持串行。

把验收点写成五组测试

Gatherer 的测试不应只检查最终列表。更有价值的是验证每一条生命周期边界：

• 空输入：不应凭空产生结果。
• 不足一批：finisher 应输出剩余元素一次。
• 恰好整批：不能额外输出空批次。
• 包含无效记录：无效项被跳过或进入错误结果，行为要与业务约定一致。
• 提前停止：integrator 返回停止信号后，不应继续产生下游结果。

@Test
void flushesTheFinalPartialBatch() {
    List result = Stream.of(a(), b(), c(), d())
        .gather(batchesOf(3))
        .toList();

    assertEquals(2, result.size());
    assertEquals(List.of(a(), b(), c()), result.get(0).members());
    assertEquals(List.of(d()), result.get(1).members());
}

接入真实业务前，再用固定输入跑一次串行和并行对照。若业务依赖顺序，测试应该明确禁止并行；若业务允许并行，则要把合并后的结果不变量写出来，而不是只断言数量相同。

常见问题

Gatherer 是不是可以替代 map 和 filter？

不必。普通的无状态转换仍然适合使用 map 和 filter。Gatherer 更适合封装带状态、需要向下游分批发出结果，或者需要提前停止的处理逻辑。

为什么示例不直接支持并行流？

因为固定窗口的成员依赖输入顺序。没有清晰的状态合并规则时，强行并行只会把问题从代码可读性变成结果不确定性。

finisher 什么时候最容易写错？

最常见的是遗漏最后一个不满批次，或者无论状态是否为空都推送一个空结果。应分别覆盖空流、整批和部分批次。

结语：先封装状态，再讨论并行

Stream Gatherer 的价值在于把“状态如何变化”和“结果何时输出”收进一个有边界的组件。先用顺序实现跑通生命周期测试，再根据状态能否拆分、结果是否允许重排来决定并行策略，通常比一开始追求更短的 Stream 表达式可靠。对批处理、窗口聚合和提前停止这类场景，清楚的状态所有权比语法压缩更重要。