Stream API 有一个很难堪的空白，从 Java 8 一直空到 Java 22：你能写有状态的中间操作，但写完肯定是错的。
写起来大概是这样：
Stream<LogEntry> stream = ...;
// 想过滤出"同一个 IP 短时间内的突发请求"
stream.filter(new Predicate<LogEntry>() {
private final Map<String, Deque<Long>> recent = new HashMap<>();
@Override
public boolean test(LogEntry entry) {
// 有状态逻辑写在这里
return false;
}
})匿名内部类持有状态，filter 里每一步都改这个状态。单线程跑得挺好，一旦有人把 stream.parallel() 加上，或者调用方不小心复用了这个 Predicate 实例，状态就串了。而且这段代码根本没办法做”输出和输入类型不同”的操作——filter 只能返回 boolean，map 只能一对一。
更别说那些真正需要自定义的东西。滑动窗口、去重、批处理、限流、去抖——在没有 Gatherer 之前，几乎所有人都是放弃 Stream，改回写循环。
Java 24 把 Stream Gatherers 正式做进来了（JEP 485）。它是一套机制，让流可以拥有真正意义上的自定义中间操作。这篇文章拿我上个月做的一个日志分析小工具来演示：从一份 Nginx 访问日志里找出短时间内的突发 IP。
版本和引入
Stream Gatherers 是 Java 24 正式版特性。Java 22 有过 preview，Java 23 第二次 preview，24 才定型。
java -version
# openjdk version "24" 2025-03-18
# 或者 25 也可以不用加 --enable-preview，标准库里就有。项目里也不用加依赖，java.util.stream.Gatherer 是 JDK 自带的。
Gatherer 接口长什么样
先把接口翻出来看一眼，理解了它就没那么玄乎。
public interface Gatherer<T, A, R> {
default Supplier<A> initializer() {
return null;
}
Integrator<A, T, R> integrator();
default BinaryOperator<A> combiner() {
return null;
}
default BiConsumer<A, Downstream<? super R>> finisher() {
return null;
}
}四个部分，前面三个几乎每个 Gatherer 都会用到。
T 是输入元素的类型，A 是内部状态类型，R 是输出元素类型。三者可以完全不一样，这也是它比 map 强大的地方。
initializer() 提供一个初始状态。如果你的 Gatherer 是无状态的（比如纯转换），返回 null 就行。
integrator() 是核心。它是一个函数接口，签名是这样的：
@FunctionalInterface
interface Integrator<A, T, R> {
boolean integrate(A state, T element, Downstream<? super R> downstream);
}每来一个元素就被调一次，你返回 true 表示”继续”，返回 false 表示”我不想再收元素了，提前结束”。
combiner() 用于并行场景，把两个中间状态合并成一个。如果不提供，即使在并行流上，这个 Gatherer 也会退化成单线程执行——这样至少保证正确。
finisher() 在流结束时被调一次。如果你有一些攒着的数据需要最后再输出（比如凑不满一批的尾部），就在这里处理。
Downstream 上只有两个方法：push(R) 往下游推一个元素，和 isRejecting() 判断下游是不是已经饱和了。前者返回 boolean，如果返回 false 通常意味着可以提前收工了。
先看内置的 Gatherers
标准库在 java.util.stream.Gatherers 类里给了五个现成的，都是平时最常自己造轮子的那种。
windowFixed — 定长分批
List<List<Integer>> batches = Stream
.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
.gather(Gatherers.windowFixed(3))
.toList();
// [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10]]注意最后一批不足 3 个也照样输出，不会丢失。
windowSliding — 滑动窗口
List<List<Integer>> windows = Stream
.of(1, 2, 3, 4, 5)
.gather(Gatherers.windowSliding(3))
.toList();
// [[1, 2, 3], [2, 3, 4], [3, 4, 5]]这个在做”移动平均”、”相邻比较”这类事情上非常顺手。以前要写好几行手工循环。
scan — 累积并输出
List<Integer> partial = Stream
.of(1, 2, 3, 4, 5)
.gather(Gatherers.scan(() -> 0, Integer::sum))
.toList();
// [1, 3, 6, 10, 15]和 reduce 的区别是 scan 每一步都往下游推，你会看到整个累积过程。做图表、做进度条、做运行总额的时候用得上。
fold — 只输出最终结果
Optional<Integer> total = Stream
.of(1, 2, 3, 4, 5)
.gather(Gatherers.fold(() -> 0, Integer::sum))
.findFirst();
// Optional[15]它和 Stream.reduce 功能重叠，但作为 gather 的组件可以在流的中途穿插使用，比如”算完小计再拿去做下一步处理”。
mapConcurrent — 并发映射
List<String> bodies = urls.stream()
.gather(Gatherers.mapConcurrent(16, this::fetch))
.toList();这个不在初版 preview 里，是后面才加的。它以最多 16 个虚拟线程并发执行 fetch，但下游接收的顺序依然和输入顺序一致。以前要做这件事得手搓 CompletableFuture 链再重排序，现在一行搞定。
案例：Nginx 日志突增检测
需求很简单。有一份 Nginx access log 解析出来的条目，每条带客户端 IP、请求时间戳、HTTP 状态码。找出在 30 秒内从同一个 IP 打进来的、超过 20 条的请求。
输出：突发的起始时间、IP、这一波里的请求数量。
先定义两个 record 作为数据模型。
public record LogEntry(
String ip,
long timestamp, // epoch millis
int status
) {}
public record Burst(
String ip,
int count,
long startTimestamp,
long detectTimestamp
) {}手工循环版本先摆出来
很多人第一反应会写成这样：
public static List<Burst> detectSequential(List<LogEntry> entries, int threshold, long windowMillis) {
Map<String, Deque<Long>> recent = new HashMap<>();
List<Burst> bursts = new ArrayList<>();
for (LogEntry entry : entries) {
Deque<Long> deque = recent.computeIfAbsent(entry.ip(), k -> new ArrayDeque<>());
long cutoff = entry.timestamp() - windowMillis;
while (!deque.isEmpty() && deque.peekFirst() < cutoff) {
deque.pollFirst();
}
int before = deque.size();
deque.addLast(entry.timestamp());
int after = deque.size();
if (before <= threshold && after > threshold) {
bursts.add(new Burst(entry.ip(), after, deque.peekFirst(), entry.timestamp()));
}
}
return bursts;
}跑起来没问题。业务逻辑也清楚：滑动窗口，每一步先淘汰过期数据，再塞新的，然后判断是不是”刚刚跨过阈值”——用 before <= threshold && after > threshold 是为了避免在同一个窗口里连续报警，只在跨过阈值的那一刻报一次。
问题是它写不成 Stream。这段逻辑的中途状态是 Map<String, Deque<Long>>，输入类型是 LogEntry，输出是 Burst——三个类型全不一样。任何一个 Stream 的内置操作都装不下这个逻辑。
而 Gatherer 装得下。
把逻辑搬进 Gatherer
状态类先单独拎出来：
final class BurstState {
final Map<String, Deque<Long>> recent = new HashMap<>();
final int threshold;
final long windowMillis;
BurstState(int threshold, long windowMillis) {
this.threshold = threshold;
this.windowMillis = windowMillis;
}
}把 threshold 和 windowMillis 放进状态里，是因为 integrator 只能拿到状态，拿不到外部的闭包变量——其实可以通过 lambda 捕获，但把它放进状态更符合 Gatherer 的语义。
然后是 integrator 的实体：
static Gatherer<LogEntry, BurstState, Burst> burstDetector(
int threshold,
long windowMillis
) {
return Gatherer.ofSequential(
() -> new BurstState(threshold, windowMillis),
(state, entry, downstream) -> {
Deque<Long> deque = state.recent.computeIfAbsent(
entry.ip(), k -> new ArrayDeque<>()
);
long cutoff = entry.timestamp() - state.windowMillis;
while (!deque.isEmpty() && deque.peekFirst() < cutoff) {
deque.pollFirst();
}
int before = deque.size();
deque.addLast(entry.timestamp());
int after = deque.size();
if (before <= state.threshold && after > state.threshold) {
downstream.push(new Burst(
entry.ip(),
after,
deque.peekFirst(),
entry.timestamp()
));
}
return true;
}
);
}Gatherer.ofSequential 是一个便捷工厂方法，专门用来造”明确规定不支持并行”的 Gatherer。因为我们内部持有 HashMap，天然就不线程安全，用它写出来语义最清楚——调用方看到这个，也不会试图去 parallel() 它。
用法：
List<LogEntry> logs = loadLogs();
List<Burst> bursts = logs.stream()
.gather(burstDetector(20, 30_000L))
.toList();
bursts.forEach(b ->
System.out.printf(
"[%d] %s 在 30 秒内请求 %d 次%n",
b.detectTimestamp(),
b.ip(),
b.count()
)
);这段代码和前面的手工循环逻辑完全一样，只是形式换了。但现在它和 Stream 的其他操作可以自由组合了：
Map<Integer, List<Burst>> byHour = logs.stream()
.filter(e -> e.status() >= 400) // 只关心错误响应
.gather(burstDetector(20, 30_000L))
.sorted(Comparator.comparingLong(Burst::detectTimestamp))
.collect(Collectors.groupingBy(
b -> (int) (b.detectTimestamp() / 3_600_000L)
));过滤、突增检测、排序、按小时分组——整条管线一气呵成。用传统的写法，这几个步骤得拆成四五个函数，中间还得定义好几个中间集合。
关于并行这件事
刚才那个 Gatherer 我用了 ofSequential，明确宣告它不支持并行。这是有原因的。
它的内部状态是 HashMap，而且逻辑本身是”全局按时间顺序扫描”。即便你提供一个 combiner 把两个 HashMap 合并起来，也会因为分区边界上的时间戳顺序错乱而算出错误结果。
换句话说，有些逻辑天然就是串行的，不要强行并行。日志突增检测就是这一类——你不知道分区 A 最后一个元素和分区 B 第一个元素在时间上是什么关系，也就无法正确地合并两个滑动窗口。
那么什么样的情况下该提供 combiner？答案是”状态可以独立累加，合并顺序不影响结果”。
举个例子：统计每个 IP 的请求总数并找出超过 1000 次的。这种就是可合并的，因为计数是交换律的。
static Gatherer<LogEntry, Map<String, Integer>, String> heavyHitters(
int threshold
) {
return Gatherer.of(
// initializer
HashMap::new,
// integrator
(state, entry, downstream) -> {
state.merge(entry.ip(), 1, Integer::sum);
return true;
},
// combiner
(left, right) -> {
right.forEach((ip, count) -> left.merge(ip, count, Integer::sum));
return left;
},
// finisher
(state, downstream) -> {
state.forEach((ip, count) -> {
if (count > threshold) {
downstream.push(ip + " → " + count);
}
});
}
);
}这个版本在 .parallel() 下可以正确完成分区分治与合并。combiner 里的 left 直接修改并返回它，是个惯用做法——避免额外的复制。
注意 finisher 的用法。它拿到最终状态，往 downstream 里吐结果。因为流结束时 downstream 还在，这才是”批量输出”的正确位置。
两个容易踩的坑
不要在 integrator 里做重活
integrator 对每个元素都会被调一次，循环里每一行代码都被放大了 N 倍。我第一版检测器里，cutoff 那行写成了 System.currentTimeMillis() - windowMillis，结果每个元素都调一次系统调用，10 万条日志跑下来慢得肉眼可见。
正确做法是用当前元素自己的时间戳——日志是有序的，每个元素都带时间，用它来算窗口边界就够了。这也是为什么前面代码里写的是 entry.timestamp() - state.windowMillis。
下游可能拒绝
downstream.push() 的返回值是 boolean。如果下游已经不需要更多了（比如调了 limit(5)，或者后面的操作短路了），它会返回 false。这时候最经济的做法是立刻 return false，终止整个 Gatherer。
if (!downstream.push(new Burst(...))) {
return false; // 下游不要了，提前收工
}downstream.isRejecting() 也可以在推之前先看一眼，但在 push 之后立刻返回 false 效果一样，少一次方法调用。
和旧写法的差异，多啰嗦几句
从我这次改写的体会来说，Gatherer 真正解决的不是”少写几行代码”，而是三件事。
第一，它是可组合的。以前这种状态ful 逻辑只能写成一个独立函数，接收 List 返回 List，中间过程看不出关系。现在它就是一个中间操作，可以和别的操作随意穿插。
第二，状态的生命周期是明确定义的。initializer 每个流实例调一次，状态归流管。匿名内部类那个坑——Predicate 被闭包捕获之后就活在别处了——从根源上消失。
第三，并行语义是显式的。ofSequential 和 of 两个工厂方法，用哪个就得承担什么后果。以前写状态ful 逻辑时的最大问题就是”这玩意能不能并行”没人说得清楚，现在只要看代码就能判断。
收个尾
试完这一圈之后，我打算把项目里几个”看起来像流、但最后写成了 for 循环”的地方逐步替换。
优先级最高的有几类：滑动窗口的指标统计、基于时间间隔的去重、批量写入前的攒批。这些场景在业务代码里到处都是，以前一般写成”接受 List 返回 List”的工具方法。换成 Gatherer 之后，可以把处理链上其他环节也保留为 Stream 操作，整条路径更一致。
门槛其实没有想象中高。接口一共四个方法，大多数场景只用两个——initializer 加 integrator。等你需要并行的时候再补 combiner 和 finisher。
如果你手上正好有一块”状态ful 过滤”或者”滑动窗口”的老代码，这是个不错的切入点。先把一个业务逻辑改成 Gatherer 跑通，再考虑要不要铺开。
