按金额切批、按字段去重这类流操作，用 Stream Gatherers 几行就能写完

Stream 的中间操作原来是固定的一套：map、filter、distinct、sorted 这些。想按固定大小分组、算滑动窗口、按某个字段去重，要么退回 for 循环，要么拿 collect 硬凑一堆临时集合。

JEP 485 在 JDK 24 把 Stream Gatherers 定稿了，之前在 JDK 22、23 预览过两轮。它给 Stream 加了一个 gather 方法，中间操作从此可以自己写。java.util.stream.Gatherers 里自带 windowFixed、windowSliding、fold、scan、mapConcurrent 五个实现，不够用就拿 Gatherer.ofSequential 或 Gatherer.of 自己拼。

下面用一个 Spring Boot 4.1 应用把这几种都跑一遍，JDK 用的是 27 GA。应用启动日志：

openjdk version "27" 2026-09-15
OpenJDK Runtime Environment (build 27+35-2325)
OpenJDK 64-Bit Server VM (build 27+35-2325, mixed mode, sharing)
08:12:12.125 INFO  DemoApplication - Started DemoApplication in 1.318 seconds (process running for 1.733)

下面每个接口的输出都来自这个进程。

先过几个内置的

windowFixed(n) 把流按每 n 个切成一组，最后不足 n 个的单独成组。攒够一批再批量写库，是最常见的用法。

@GetMapping("/window-fixed")
public List<List<Integer>> windowFixed() {
    return Stream.of(10, 20, 30, 40, 50, 60, 70)
            .gather(Gatherers.windowFixed(3))
            .toList();
}

$ curl -s localhost:18088/window-fixed
[[10,20,30],[40,50,60],[70]]

windowSliding(n) 每次只往前挪一格，相邻两组重叠 n-1 个。算移动平均直接套上去。

@GetMapping("/moving-average")
public List<Double> movingAverage() {
    return Stream.of(10, 20, 30, 40, 50)
            .gather(Gatherers.windowSliding(3))
            .map(w -> w.stream().mapToInt(Integer::intValue).average().orElse(0))
            .toList();
}

$ curl -s localhost:18088/moving-average
[20.0,30.0,40.0]

scan 和 reduce 不一样，它把每一步的中间结果都发下去，不是只给最后一个值。做运行总和、逐步累积状态正合适。

@GetMapping("/running-sum")
public List<Integer> runningSum() {
    return Stream.of(1, 2, 3, 4, 5)
            .gather(Gatherers.scan(() -> 0, Integer::sum))
            .toList();
}

$ curl -s localhost:18088/running-sum
[1,3,6,10,15]

fold 可以看成 scan 的单值版，只在流结束时发出一个累积结果，这里不单独演示。mapConcurrent 用虚拟线程并发执行映射函数，可以限定最大并发数，属于另一类用法，这里也不展开。

自己写一个带状态的 Gatherer

自定义才是重点。Gatherer.ofSequential 有几个重载，参数最全的一版收三样：initializer 提供初始状态，Integrator 处理每个元素，finisher 在流结束时收尾。initializer 和 finisher 都可以省。名字里的 Sequential 表示它只按顺序执行。要支持并行流，得用 Gatherer.of，再多给一个 combiner 合并各段的状态。

先看按 key 去重。状态就是一个已见过的 key 集合。

public static <T, K> Gatherer<T, ?, T> distinctBy(Function<? super T, ? extends K> keyFn) {
    return Gatherer.ofSequential(
            () -> new java.util.HashSet<K>(),
            (seen, element, downstream) -> {
                if (seen.add(keyFn.apply(element))) {
                    return downstream.push(element);
                }
                return true;
            });
}

integrator 的第三个参数 downstream 就是下游。push 返回 false 表示下游不再接收元素。把这个返回值原样返回，流就能提前停下。标准库的 Stream.distinct 只能按 equals 去重，按某个字段去重以前得自己想办法，现在写成一个 gatherer 就能到处复用。

@GetMapping("/distinct-by")
public List<Reading> distinctBy() {
    return Stream.of(
                    new Reading("a", 1), new Reading("b", 2),
                    new Reading("a", 3), new Reading("c", 4),
                    new Reading("b", 5))
            .gather(Gatherers485.distinctBy(Reading::sensor))
            .toList();
}

$ curl -s localhost:18088/distinct-by
[{"sensor":"a","value":1},{"sensor":"b","value":2},{"sensor":"c","value":4}]

a 和 b 的第二条都被丢掉了，每个传感器只留第一条。

带收尾逻辑的分批

windowFixed 只能按个数切。要按累计金额切批，就得自己攒状态。下面这个累计到 100 就切一批，剩下不满 100 的，在流结束时由 finisher 补发出去。

public static Gatherer<Integer, ?, List<Integer>> batchUntilSum(int limit) {
    class Acc {
        final List<Integer> items = new ArrayList<>();
        int sum;
    }
    return Gatherer.ofSequential(
            Acc::new,
            (acc, element, downstream) -> {
                acc.items.add(element);
                acc.sum += element;
                if (acc.sum >= limit) {
                    boolean ok = downstream.push(List.copyOf(acc.items));
                    acc.items.clear();
                    acc.sum = 0;
                    return ok;
                }
                return true;
            },
            (acc, downstream) -> {
                if (!acc.items.isEmpty()) {
                    downstream.push(List.copyOf(acc.items));
                }
            });
}

第三个 lambda 就是 finisher。没有它，最后那批不满 100 的元素会被悄悄吞掉。

$ curl -s localhost:18088/batch-until-sum
[[30,40,50],[10,5,90],[20]]

30+40+50 到 120，切一批。10+5+90 到 105，再切一批。最后剩下的 20 由 finisher 补上。

gatherer 可以串起来

gather 后面能接着 gather，和普通中间操作一样往下链。下面先按传感器去重，再每 2 个一组。想把两个 gatherer 合成一个复用，也可以用 Gatherer.andThen。

@GetMapping("/compose")
public Map<String, Object> compose() {
    var result = Stream.of(
                    new Reading("a", 1), new Reading("a", 9),
                    new Reading("b", 2), new Reading("c", 3),
                    new Reading("c", 8), new Reading("d", 4))
            .gather(Gatherers485.distinctBy(Reading::sensor))
            .gather(Gatherers.windowFixed(2))
            .toList();
    Map<String, Object> out = new LinkedHashMap<>();
    out.put("jdk", Runtime.version().toString());
    out.put("groups", result);
    return out;
}

$ curl -s localhost:18088/compose
{"jdk":"27+35-2325","groups":[[{"sensor":"a","value":1},{"sensor":"b","value":2}],[{"sensor":"c","value":3},{"sensor":"d","value":4}]]}

去重后剩 a、b、c、d 四条，再两两成组。

和 mapMulti 的分工

之前那篇不要使用 Stream 的 filter().map()，试试 mapMulti()讲的是一对多展开：一个元素摊成零到多个，元素之间不共享状态。Gatherers 管的是另一类事：处理当前元素之前，得记住前面看过什么，比如分组、去重、累积、滑窗。两者各管一摊，并不冲突。