上个月把一个对账任务从 JDK 17 升到 JDK 24，最想动的地方不是虚拟线程，而是那段用 for 循环加临时 List 拼出来的批处理逻辑。JDK 24 正式把 Stream Gatherers 转正（JEP 485），java.util.stream.Gatherers 可以直接用，不用再加 --enable-preview。这篇文章按我当时重构的顺序写：先看旧代码怎么写的，再看 Gatherer 的四个回调分别负责什么，最后把滑动窗口、批量落库、并发拉取价格的代码拼成一条流水线。所有代码在 JDK 24 上直接编译运行。
一、重构前：三段嵌套的批处理
任务是从订单流里每 100 条写一次数据库，最后不足 100 条的尾巴也要写：
List<List<Order>> batches = new ArrayList<>();
List<Order> buffer = new ArrayList<>();
for (Order order : orders) {
buffer.add(order);
if (buffer.size() == 100) {
batches.add(buffer);
buffer = new ArrayList<>();
}
}
if (!buffer.isEmpty()) {
batches.add(buffer);
}
for (List<Order> batch : batches) {
dao.saveBatch(batch);
}能跑，但问题明显：遍历和分批耦合；尾巴单独处理；想加个“每 5 条算一次移动平均”就得再写一个循环。Stream 的 map / filter 处理不了这种“多个元素合成一个”或者“跨元素记住状态”的场景，collect 又只能放在最后，中间不能继续接 filter。Gatherer 就是填这个空档的。
二、Gatherer 的四个回调，各自管什么
一个 Gatherer<T, A, R> 描述的是“把 T 类型的元素，用 A 类型的状态，转换成 R 类型”的中间操作。四个可以覆写的方法：
initializer()：返回状态的初始值。每份独立的状态（并行时每片一个）都靠它创建。无状态 gatherer 返回 null 就行。integrator()：核心。每来一个元素调用一次，可以向下游 push 0 个、1 个或多个结果。返回 false 表示“我不想再收元素了”，上游会停机。combiner()：并行流里合并两份状态用。顺序流不需要，返回 null。finisher()：流结束时调用一次，用来吐出最后攒着的结果。上面那段旧代码里的尾巴就归它管。
和 Collector 的区别一句话：Collector 是终端操作，用完流就结束了；Gatherer 是中间操作，结果仍然是一个 Stream，后面还能继续 map、filter、再 gather。
三、内置的五个 Gatherer，够应付八成场景
JDK 自带的 Gatherers 工厂方法不多，但都很实用。先跑通一遍找找感觉。
windowFixed：定长切块
Stream.of(1, 2, 3, 4, 5, 6, 7)
.gather(Gatherers.windowFixed(3))
.forEach(System.out::println);
// [1, 2, 3]
// [4, 5, 6]
// [7]直接拿来替掉前面的 for 循环：
orders.stream()
.gather(Gatherers.windowFixed(100))
.forEach(dao::saveBatch);尾巴那一批由 windowFixed 自己处理，不满 100 条也会下发。唯一要注意的是每个窗口是新的容器，不要想着复用。
windowSliding：滑动窗口
Stream.of(120, 80, 300, 90, 60)
.gather(Gatherers.windowSliding(3))
.forEach(System.out::println);
// [120, 80, 300]
// [80, 300, 90]
// [300, 90, 60]做移动平均、连续三次失败告警这类需求，比手写 subList 省心。窗口大小如果大于元素个数，什么都不会输出。
scan 和 fold：一个每次产出，一个最后产出
这两个名字容易混。scan 是“带历史的 map”，每处理一个元素就产出一个累积值：
Stream.of(100, -30, 20, -150)
.gather(Gatherers.scan(() -> 0, Integer::sum))
.forEach(System.out::println);
// 100
// 70
// 90
// -60正好是账户流水的余额变化。fold 则只在流结束时产出一个值，相当于把 reduce 搬到了中间位置，后面还能接着处理：
Optional<Integer> total = Stream.of(100, -30, 20)
.gather(Gatherers.fold(() -> 0, Integer::sum))
.findFirst();
System.out.println(total); // Optional[90]如果数据源是无限的，scan 能用，fold 用不了——它得等流结束。
mapConcurrent：并发映射，顺序不变
List<String> skus = List.of("A100", "B200", "C300", "D400");
List<Quote> quotes = skus.stream()
.gather(Gatherers.mapConcurrent(8, sku -> remoteQuote(sku)))
.toList();maxConcurrency 是同时在跑的任务数上限。它和 parallelStream() 不是一回事：这里不需要流本身是并行的，即使顺序流也会并发调用 mapper，而且下游拿到的结果顺序和输入顺序一致。mapper 会被多个线程同时调用，里面别写共享可变状态，远程调用也要自己在外部限流。
四、自己写一个 batch Gatherer
内置的 windowFixed 已经能满足大部分分批需求，但它没法控制“每批之间的间隔”或者“按大小和内容混合触发”。自己写一个反而更能看清三个回调怎么配合：
static <T> Gatherer<T, ?, List<T>> batch(int size) {
return Gatherer.<T, List<T>, List<T>>ofSequential(
() -> new ArrayList<>(),
(List<T> buf, T element, Gatherer.Downstream<? super List<T>> downstream) -> {
buf.add(element);
if (buf.size() == size) {
List<T> snapshot = List.copyOf(buf);
buf.clear();
return downstream.push(snapshot);
}
return true;
},
(List<T> buf, Gatherer.Downstream<List<T>> downstream) -> {
if (!buf.isEmpty()) {
downstream.push(List.copyOf(buf));
}
});
}三件事对应三个 lambda：初始状态是一个空 List；每来一个元素就塞进去，满了就复制一份推给下游并清空；流结束时把剩下的推出去。注意这里用的是 List.copyOf 而不是直接 push(buf)，因为清空之后 buf 还会接着复用，直接推出去的话下游拿到的都是同一个空 List。
用法和内置的没区别：
orders.stream()
.filter(Order::valid)
.gather(batch(100))
.forEach(dao::saveBatch);如果需要滚动窗口的统计，也可以换成自定义的移动平均：
static Gatherer<Integer, ?, Double> movingAverage(int windowSize) {
return Gatherer.<Integer, Deque<Integer>, Double>ofSequential(
ArrayDeque::new,
(Deque<Integer> window, Integer element,
Gatherer.Downstream<? super Double> downstream) -> {
window.addLast(element);
if (window.size() > windowSize) {
window.removeFirst();
}
if (window.size() < windowSize) {
return true;
}
double avg = window.stream()
.mapToInt(Integer::intValue)
.average()
.orElseThrow();
return downstream.push(avg);
});
}Stream.of(120, 80, 300, 90, 60, 1000, 40)
.gather(movingAverage(3))
.forEach(avg -> System.out.printf("%.1f%n", avg));
// 166.7
// 156.7
// 150.0
// 383.3
// 366.7这里没用 finisher，因为窗口不满就直接不输出，没有尾巴要处理。用到 ofSequential 是因为窗口顺序对结果有影响，不希望它在并行流里被拆开合并。
五、让 Gatherer 主动停机
integrator 返回 false 是一个容易被忽略的能力。它不等于 limit(n)，而是“根据已经处理过的数据做判断，随时收工”。比如统计响应时间，累计超过 1 秒就不再往下走：
static Gatherer<Integer, ?, Integer> takeUntilTotalCost(int limitMs) {
return Gatherer.<Integer, int[], Integer>ofSequential(
() -> new int[1],
(int[] total, Integer cost,
Gatherer.Downstream<? super Integer> downstream) -> {
total[0] += cost;
boolean accepted = downstream.push(cost);
return accepted && total[0] < limitMs;
});
}Stream.of(120, 80, 300, 90, 1000, 5000)
.gather(takeUntilTotalCost(1000))
.forEach(c -> System.out.print(c + " "));
// 120 80 300 90 1000累计到 1590 的时候，下一次调用返回 false，上游不再继续产生元素。这个能力在处理大文件、长轮询、埋点采样时特别顺手。
六、拼一条完整的流水线
把上面的东西组合起来，模拟一段接口日志的处理：过滤掉解析失败的行，用响应时间做 3 期移动平均，每 3 个平均值打一批点。前面两个静态方法放进同一个类里即可：
public class LogPipeline {
record AccessLog(String path, int costMs) {}
public static void main(String[] args) {
List<AccessLog> logs = List.of(
new AccessLog("/api/order", 120),
new AccessLog("/api/order", 80),
new AccessLog("/api/user", 300),
new AccessLog("/api/order", 90),
new AccessLog("/api/user", 60),
new AccessLog("/api/pay", 1000),
new AccessLog("/api/pay", 40),
new AccessLog("/api/pay", 200),
new AccessLog("/api/order", 150),
new AccessLog("/api/order", 70));
logs.stream()
.filter(log -> log.costMs() > 0)
.map(AccessLog::costMs)
.gather(movingAverage(3))
.map(avg -> Math.round(avg))
.gather(batch(3))
.forEach(batch -> System.out.println("上报一批指标: " + batch));
}
}输出是三批：前两批各 3 个平均值，最后一批 2 个。整条链路是惰性的，直到 forEach 才开始跑。
顺便说一下，这里出现了两次 gather。Gatherer 是普通中间操作，可以在流里出现多次，这一点比把逻辑全塞进一个 Collector 要灵活。
七、几个实际踩到的坑
- finisher 忘了写，最后一批数据就没了。凡是“攒够才输出”的 gatherer，都要检查尾巴。
- 把可变的容器直接 push 给下游，然后自己清空继续用，下游拿到的内容会在之后被改掉。要么
List.copyOf，要么每次 new 一个新容器。 - 有状态的 gatherer 不是线程安全的。用
of声明并行能力的时候，combiner 必须真的能合并状态，而且要满足结合律，否则并行流下结果不稳定。没把握就用ofSequential，然后在顺序流上跑。 windowSliding每个窗口都会生成一个 List，窗口开得太大、元素又多的时候，GC 压力比手写环形数组要明显。数据量上了百万级别，先压测再上。- 不要在 integrator 里做阻塞的远程调用。需要并发就用
mapConcurrent，把 io 放在 mapper 里。 mapConcurrent的 maxConcurrency 传 0 或负数会直接抛IllegalArgumentException，配置项最好在启动时校验一遍。
八、什么时候该用它
判断标准很简单：如果一段流式代码需要“记住前面几个元素”、“攒够 N 个再输出”、“按业务条件提前结束”，而现有中间操作写起来别扭，那就是 Gatherer 的用武之地。反过来，单纯的 1 对 1 转换继续用 map，能过滤就用 filter，最后要聚合成一个结果还是 collect 更合适。
迁移的时候不用一次改完。先把项目升到 JDK 24，确认 java -version 输出 24，然后在最别扭的那段循环上下手，用内置的 windowFixed 或者 scan 试一次，感受一下短路和 finisher 的配合，再决定要不要抽成通用方法。
