当前位置：首页 > 文章列表 > 文章 > java教程 > Java Gatherers.fold 如何做分段聚合：状态边界与并行流验收

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

来源：17golang原创 2026-08-28 02:31:39 0浏览收藏

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

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

实践要点

Gatherers.fold 用于没有可实现 combiner 的有序聚合。
Supplier 返回新的聚合状态，避免不同流实例共享同一个可变对象。
空流会保留初始状态，非空流在正常结束时只产生一个结果。
并行流先看结果顺序和状态隔离，再谈吞吐；本例默认用串行流。

一次错误改造：把有序摘要当成普通 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 塞给所有流实例，所以每次调用都有自己的起点。

StringBuilder::new 创建状态，order.id() 写入元素，findFirst 取出唯一摘要，orElseGet 处理空流

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

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");
}