上个月在做订单报表模块的时候，有一段逻辑让我纠结了好几天。需求是：从数据库里查出最近三个月的原始订单流，按天分组统计每类商品的销售额，同时要求输出一个每10分钟时间窗口的订单量滑动平均值，再加一个特殊规则——如果某个商品的单笔金额超过阈值，需要跳过该记录并启动一个“异常累计计数器”，当计数器达到5时自动截断窗口。用现有的 Stream API 搭配 Collectors 可以实现，但代码里不得不塞进一堆 `flatMap`、`collect(Collectors.groupingBy(…))` 和自定义的 Collector 实现，最后整个方法体超过了 80 行，可读性非常差。
当时正好看到 Java 24 的早期构建里包含了一个预览特性——Stream Gatherers（JEP 485），它的目标就是让开发者可以自定义流中间操作，就像 `map` 或 `filter` 一样，但能够携带状态并控制输出的时机。这正好对应了我那几个“有状态的、需要累积再一并输出”的需求。花了两个晚上把那段 80 行的流式处理用 Gatherers 重写之后，代码变成了 30 多行，而且每个处理步骤的逻辑都收拢到了一个独立的 Gatherer 里。这篇文章就把这个过程和方法完整拆解出来。
一、传统 Stream 在复杂数据处理上的局限
Stream API 自从 Java 8 引入以来，一直是数据流水线处理的首选。但对于有状态的、需要累积后再输出的场景，它的原生支持一直比较薄弱。比如要按固定大小分组（不是按谓词，而是每 N 个元素一组），你只能通过索引操作或者用 `Collectors` 配合 `ArrayList` 自己攒一个自定义收集器。再比如需要根据当前累积的状态决定是输出还是跳过，这种逻辑在 `map` 和 `filter` 里做不了，因为它们都是无状态的。
之前我为了完成那个“每 10 分钟窗口滑动平均”的需求，写了这样一个自定义 Collector：
Collector<Order, ?, List<Double>> slidingWindowCollector = Collector.of(
ArrayDeque::new,
(deque, order) -> {
deque.addLast(order.getAmount());
if (deque.size() > WINDOW_SIZE) {
deque.removeFirst();
}
},
(d1, d2) -> { d1.addAll(d2); return d1; },
deque -> deque.stream().mapToDouble(Double::doubleValue).average().stream()
.mapToObj(avg -> List.of(avg)).orElse(List.of())
);这段代码不仅隐晦，而且因为中间状态是 `ArrayDeque` 这种非线程安全的容器，稍有不慎就会在并行流中出错。更要命的是，这个 Collector 是终止操作，不能嵌入到流的中间，一旦用了它，后续就不能再继续做其他流操作了。所以实际使用时还得把结果包装成新流再次处理，代码变得割裂。
二、Stream Gatherers 的核心概念
Gatherers 为 Stream API 增加了一种新的中间操作类型，由 `Stream.gather(Gatherer)` 方法提供入口。一个 Gatherer 可以随时决定是输出元素、跳过元素、还是暂存状态并在合适的时机发出累积结果。它由四个核心组件构成：
- 初始状态工厂（initializer）：提供一个可变的状态容器，在每次流计算开始时调用。
- 整合器（integrator）：接收当前状态、下一个流元素和下游输出器，决定是否输出、更新状态、并返回一个布尔值指示是否继续处理。
- 合并器（combiner）：用于并行流中合并两个状态。
- 终结器（finisher）：在流结束时调用，用于输出残留的状态。
这和 Collector 的接口非常相似，但 Gatherer 工作在流的中间，可以像 `map` 一样继续拼接后续操作。这意味着你可以把复杂的有状态逻辑封装成一个 Gatherer，然后把它插在 `filter` 和 `map` 之间，不再需要终止流再重新开启。
三、案例一：按订单状态分组并计算总金额
先从一个相对简单的需求切入：将原始订单流按状态（如“已支付”、“待发货”等）分组，并实时输出每组的累计金额。如果使用 `Collectors.groupingBy` 加 `summingDouble`，必须在流终止时一次性拿到全部结果，无法在累积过程中看到中间快照。而用 Gatherer 可以做到每当一组金额累加变化时就向后续步骤输出一条汇总记录。
定义一个 `GroupAggregator` Gatherer：
Gatherer<Order, Map<String, Double>, OrderSummary> groupAggregator = Gatherer.ofSequential(
// 初始状态：一个空的 HashMap
HashMap::new,
// 整合器：接收 Map 状态、当前订单、下游
(state, order, downstream) -> {
state.merge(order.getStatus(), order.getAmount(), Double::sum);
// 每更新一次就输出当前分组汇总
state.forEach((status, sum) ->
downstream.push(new OrderSummary(status, sum)));
return true; // 继续处理下一个元素
},
// 终结器：流结束时再输出一次最终结果，也可以不输出
(state, downstream) -> {
state.forEach((status, sum) ->
downstream.push(new OrderSummary(status, sum)));
}
);使用它：
List<OrderSummary> results = orders.stream()
.gather(groupAggregator)
.toList();比起原来需要先 `collect(Collectors.groupingBy(…))` 再单独循环遍历 Map 转换成列表，现在整个逻辑是一条连贯的流式调用。而且因为 Gatherer 是中间操作，我们还能在它后面加 `.filter(s -> s.total() > 1000)` 只输出大额分组，这在 Collector 方案里需要额外步骤。
四、案例二：固定大小滑动窗口计算移动平均
回到前面提到的滑动窗口需求。我们用 Gatherer 来实现每 10 个订单作为一个窗口计算平均金额，并滑动步长为 1。整合器中维护一个 `ArrayDeque`，当队列大小达到窗口大小时，计算平均值并输出，然后移除最早的元素以保持窗口大小。
Gatherer<Order, ArrayDeque<Double>, Double> slidingAvg = Gatherer.ofSequential(
ArrayDeque::new,
(deque, order, downstream) -> {
deque.addLast(order.getAmount());
if (deque.size() == 10) {
double avg = deque.stream().mapToDouble(Double::doubleValue).average().orElse(0);
downstream.push(avg);
deque.removeFirst();
}
return true;
},
(deque, downstream) -> {
if (!deque.isEmpty()) {
double avg = deque.stream().mapToDouble(Double::doubleValue).average().orElse(0);
downstream.push(avg);
}
}
);调用方式：
List<Double> movingAverages = orders.stream()
.gather(slidingAvg)
.toList();与之前自定义 Collector 的版本相比，这个 Gatherer 直接嵌在流中，后面还可以继续做 `map` 或 `filter`。而且代码意图非常清晰：状态是一个队列，逻辑就是添加/移除/计算平均，完全没有 Collector 里那些别扭的 `finisher` 返回值的困扰。
五、案例三：有状态过滤——超阈值跳过与异常累计
最后一个需求是当订单金额超过阈值（比如 50000 元）时，跳过该订单，但对一个“异常计数器”加 1。一旦计数器达到 5，立即截断窗口，将当前累积的结果输出并清空状态。这种逻辑在普通 Stream 中几乎无法实现，因为它需要根据历史状态改变后续元素的行为。
用 Gatherer 实现它：状态包含当前窗口的订单列表和一个异常计数器。整合器判断金额是否超阈值，如果是则递增计数器但不加入列表；如果计数器达到 5，则输出当前列表并清空。普通订单正常加入列表。
Gatherer<Order, WindowState, List<Order>> thresholdGatherer = Gatherer.ofSequential(
() -> new WindowState(new ArrayList(), 0),
(state, order, downstream) -> {
if (order.getAmount() > 50000) {
state.abnormalCount++;
if (state.abnormalCount >= 5) {
// 输出当前窗口并重置
if (!state.buffer.isEmpty()) {
downstream.push(new ArrayList(state.buffer));
state.buffer.clear();
}
state.abnormalCount = 0;
}
return true;
}
state.buffer.add(order);
return true;
},
(state, downstream) -> {
if (!state.buffer.isEmpty()) {
downstream.push(new ArrayList(state.buffer));
}
}
);
record WindowState(List<Order> buffer, int abnormalCount) {}这种复杂的状态控制如果用传统 Collectors 来实现，可能需要借助 `AtomicInteger` 和手动循环，完全失去声明式风格。而 Gatherer 把它封装得像是流上的一个原生步骤。
六、与 Collector 的本质区别及选用建议
Collector 是终止操作，它消费整个流并产出最终结果；Gatherer 是中间操作，它可以随时输出、过滤或转换元素，且流可以继续。当你需要“边处理边输出”或者“输出多个元素”的时候，Gatherer 明显更合适。此外，Gatherer 可以集成到现有流水线中，支持短路操作（返回 false 停止处理），并且可以处理无限流，而 Collector 只能在全量数据上操作。
目前 Gatherers 是预览特性，使用时需要添加 `–enable-preview` 参数。Java 24 中引入的是一组工厂方法（`Gatherer.ofSequential`、`Gatherer.of` 等），以及几个内置的 Gatherer，如 `Gatherers.windowFixed`、`Gatherers.windowSliding`、`Gatherers.fold`、`Gatherers.concat` 等，可以直接使用而不必全部自己写。
七、总结
把报表模块从 Collector 堆砌改为 Gatherers 之后，代码量减少了将近三分之二，每个 Gatherer 都可以独立复用和测试。更关键的是，整条流式管道从数据源到最终输出一气呵成，不再被 `collect` 截断成几段。对于经常和数据流打交道的后端开发来说，Gatherers 这种能自定义中间操作的能力，让 Stream API 的适用范围从简单的映射过滤一下子扩展到了复杂的有状态场景。尽管目前还是预览阶段，但它的特性和设计已经相当成熟，完全值得在新项目里提前了解和试用。
以后碰到类似的分组聚合、滑动窗口、带条件过滤的需求，第一个念头不再是如何拼凑那些拗口的 Collector 签个名，而是直接打开 Gatherer 的文档，把状态逻辑往整合器里一填，剩下的交给流自己跑——这种感觉，很舒服。
