Java Gatherers.fold 如何做分段聚合：状态边界与并行流验收

线上账单摘要有个很小但容易误用的需求：订单要按到达顺序拼成一行，最后再交给审计日志。Java 24 的 Gatherers.fold 正好适合这种“只有一个结果、而且每一步都依赖上一步状态”的处理；它不等于可以随手改成并行的 reduce。

把 fold 看成有序状态机：initial 提供起点，folder 一次接收一个元素；空流也会产出初始值，无法证明顺序安全时就保持串行验收。

一次错误改造：把有序摘要当成普通 reduce

问题出在账单导出器的一次“小优化”。原逻辑把每笔订单写进同一个 StringBuilder，改造后有人想用并行流提速，再用字符串拼接收尾。结果不是稳定变快，而是验收样本的顺序偶尔变化，空批次也被误判成“没有结果”。

这类任务的关键不是 API 长什么样，而是聚合状态是否能被安全拆分、合并。Oracle 的 Java SE 24 API 将 fold 定义为有序、类似归约的转换，特别适用于无法实现合并函数或本质依赖顺序的情况。

先把状态边界写进最小示例

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

record Order(String id, int cents) {}

static String summarize(List orders) {
    Optional result = orders.stream()
        .gather(Gatherers.fold(
            StringBuilder::new,
            (summary, order) -> {
                if (!summary.isEmpty()) summary.append('|');
                summary.append(order.id()).append('=').append(order.cents());
                return summary;
            }))
        .findFirst();
    return result.orElseGet(StringBuilder::new).toString();
}

var text = summarize(List.of(
    new Order("A17", 1200),
    new Order("B03", 980)
));
// A17=1200|B03=980

这里的正文节点是 StringBuilder::new、order.id()、findFirst 和 orElseGet。它们分别对应状态创建、当前元素写入、唯一结果取出和空结果兜底。代码没有把一个外部可变的 StringBuilder 塞给所有流实例，所以每次调用都有自己的起点。

触发条件：空流和终止时机必须单独验收

fold 不是“每来一条就向下游发一条”的扫描操作。官方实现契约指出，处理过程没有抛出异常时，它只会产生一个元素；因此示例先用 findFirst 取出最终状态，再转成字符串。

空输入也值得写测试。StringBuilder::new 会创建空状态，orElseGet 最终得到空字符串。不要把空字符串和“没有执行过聚合”混在业务语义里；如果审计日志需要区分两者，就在返回类型中增加明确状态，而不是猜测 Optional。

var empty = summarize(List.of());
if (!empty.isEmpty()) {
    throw new IllegalStateException("empty batch changed");
}

var ordered = summarize(List.of(
    new Order("A17", 1200),
    new Order("B03", 980)
));
if (!ordered.equals("A17=1200|B03=980")) {
    throw new IllegalStateException("order changed: " + ordered);
}

修复动作：把并行流放进对照实验

Java 24 的 Stream.gather 支持有状态中间操作，文档也说明并行执行时可能建立多个中间结果并进行合并。但这条能力描述不能替我们证明某个业务状态适合并行；本例的拼接顺序和最终文本都依赖前一步，因此生产路径保持串行。

验收时可以保留一组对照数据，用来及时发现调用方偷偷改成并行流：

var expected = summarize(List.of(
    new Order("A17", 1200),
    new Order("B03", 980)
));

var parallelObserved = List.of(
    new Order("A17", 1200),
    new Order("B03", 980)
).parallelStream()
 .gather(Gatherers.fold(
     StringBuilder::new,
     (summary, order) -> {
         if (!summary.isEmpty()) summary.append('|');
         return summary.append(order.id()).append('=').append(order.cents());
     }))
 .findFirst()
 .map(StringBuilder::toString)
 .orElse("");

if (!expected.equals(parallelObserved)) {
    throw new IllegalStateException("parallel order changed");
}

这个对照实验的可见结果很简单：串行结果必须是 A17=1200|B03=980。如果并行对照不稳定，就停止“提速”改造；如果样本稳定，也仍要结合真实数据量和顺序契约继续测量，不能仅凭一次样本宣布安全。

防复发清单：什么时候换回 reduce

• 每个分片都能独立累加，且合并满足结合律：优先评估标准 reduce 或 collect。
• 聚合结果必须按输入先后拼接，且没有可靠的 combiner：保留 Gatherers.fold 和串行流。
• 初始值是可变对象：使用 supplier 每次创建新实例，不要复用静态 StringBuilder。
• 需要观察中间结果：不要把 fold 误替换成 scan，两者输出契约不同。

相关问题

Gatherers.fold 空流会返回什么？

正常结束时会由 initial 提供一个结果，所以示例里的空流得到空的 StringBuilder 状态。

为什么示例还调用 findFirst？

因为 fold 的正常处理结果只有一个元素，findFirst 把这个唯一状态取出来，代码语义也比收集成列表更直接。

有了并行流对照通过就能上线吗？

不能。还要证明状态可隔离、顺序契约可接受，并用真实批量规模测量；本例的顺序依赖仍然让串行路径更稳妥。

最后的验收结果

这次修复没有追求把每个流都并行化，而是先把状态起点、元素写入、唯一结果和空流分支写清楚。对 Java 24 的 Gatherers.fold 来说，能解释为什么必须按顺序处理，往往比多出一个线程更接近正确答案。