Java Stream Gatherers：从内置模式到自定义管道机制

1. 引言与背景
2. 入门：内置 Gatherers 实战
- 2.1 理解流的基数
- 2.2 五个开箱即用的 Gatherers
3. 进阶：从使用 Gatherers 到编写自己的 Gatherers
- 3.1 何时应编写自定义 Gatherer？
- 3.2 使用 andThen() 进行管道组合
4. 深入底层：自定义 Gatherers 的架构与剖析
5. 生产级 Gatherers：并行、状态与陷阱
6. 结论
参考文献

[LOADING...]

1. 引言与背景

1.1 Streams 的演进

在之前的文章中，我已经介绍了 Streams API 的演进。

在本文中，我们将深入探讨 Java 的 Streams Gatherers 特性，并说明它为开发者带来的益处。该特性有助于创建自定义操作，并支持与 map、filter、reduce 等内置操作同时进行数据转换。

1.2 “Collector 问题”

作为 Java 开发者，我们在处理复杂的嵌套 Collector 和多步骤操作时经常会遇到挑战。编写这类代码往往会导致代码冗长且复杂。为了解决这个问题，我们可以利用创建自定义中间操作的新能力。

1.3 什么是 Stream Gatherers？

Stream Gatherers 是一项新能力，最初由 JEP 461 - Streams Gatherers 作为预览特性引入，并在 JDK 24 中通过 JEP 485 最终确定。它支持创建自定义中间操作，为在流管道中转换数据提供了灵活性，而这些转换方式往往难以通过现有内置中间操作直接实现。

2. 入门：内置 Gatherers 实战

2.1 理解流的基数

在 Java Stream Gatherers 引入之前，Stream API 中的每个中间操作都受严格的“基数规则”约束，这些规则定义了进入操作的元素数量与离开操作的元素数量之间的数学关系。

为了说明 Stream Gatherers 带来的变革性影响，让我们看看标准中间操作如何管理基数：

1 对 1（map）：每个输入元素经过转换后恰好产生一个输出元素。
- 示例：将字符串流转为小写或大写，并获取其长度 stream.map(String::length) 或 stream.map(String::toUpperCase)
1 对 0 或 1（filter）：每个输入元素最多产生一个输出元素（要么通过，要么被丢弃）
- 示例：仅保留偶数 stream.filter(n -> n % 2 == 0)
1 对多（flatMap）：每个输入元素可以生成零个、一个或多个输出元素，这些元素随后会被展平为连续的流。
- 示例：将句子流拆分为单个单词（stream.flatMap(sentence -> Arrays.stream(sentence.split(" ")))）。

虽然大多数内置中间操作覆盖了日常数据转换的绝大多数场景，但它们有一个显著限制：它们是无状态的，并且独立地对每个元素进行操作。当业务逻辑需要多对多关系或跨多个元素的有状态分组时，例如：

将项目分组为固定大小为 10 的批次，以便批量插入数据库
计算金融股票价格的滑动窗口移动平均值
将元素与其前一个或后一个元素进行比较
运行累计求和或滚动余额

大多数开发者会求助于嵌套 Collectors、Map、Transform，或过度使用 Collectors，这会导致代码冗长。

Stream gatherers 正好弥合了这一缺口，使中间操作能够维护私有状态、缓冲元素，并沿管道向下游发出自定义数据块——同时保持流干净、惰性且函数式纯粹。

2.2 五个开箱即用的 Gatherers

java.util.stream.Gatherers 是一个工厂类，用于为 Java Stream API 提供自定义中间操作的标准内置实现。

最近，我通过使用 HackerNews 的端点获取热门故事，探索了虚拟线程。在这次探索中，我尝试实现了内置 Gatherers 方法。

2.2.1 使用 `windowFixed` 批量处理元素

Gatherers.windowFixed(size)：将传入的流元素拆分为指定最大大小的不重叠列表（批次）。如果流大小不能被整除，最后一批可能包含更少元素。

使用场景： 模拟 HackerNews 热门故事 ID，或将项目分批用于批量数据库插入、分页块或批量获取 API 请求。

$ java

// 模拟 Hacker News 热门故事 ID
List<Integer> storyIds = List.of(50028275, 50029123, 50019911, 50022292, 49997481, 49981264, 50023450);

// 分组为固定大小为 3 的窗口
List<List<Integer>> batches = storyIds.stream()
    .gather(Gatherers.windowFixed(3))
    .toList();

batches.forEach(batch -> IO.println("Batch: " + batch));
// 输出：
// Batch: [50028275, 50029123, 50019911]
// Batch: [50022292, 49997481, 49981264]
// Batch: [50023450]

2.2.2 使用 `windowSliding` 分析序列

Gatherers.windowSliding(size)：生成预定大小的滑动窗口。与 windowFixed 不同，这些窗口会重叠，每次向前移动一个元素。

使用场景： 利用相邻故事 ID 的滑动窗口来监控提交速度，或实时识别顺序错乱。

$ java

List<Integer> storyIds = List.of(50028275, 50029123, 50019911, 50022292, 49997481, 49981264, 50023450);

// 检查 3 个连续故事 ID 的滑动窗口
storyIds.stream()
    .limit(10)
    .gather(Gatherers.windowSliding(3))
    .forEach(window -> IO.println("Sliding ID Window: " + window));

// 输出：
Sliding ID Window: [50028275, 50029123, 50019911]
Sliding ID Window: [50029123, 50019911, 50022292]
Sliding ID Window: [50019911, 50022292, 49997481]
Sliding ID Window: [50022292, 49997481, 49981264]
Sliding ID Window: [49997481, 49981264, 50023450]

2.2.3 使用 `scan` 累积值

Gatherers.scan(initial, function)：它会像累计总额一样持续累加数值，并沿途输出每一步的结果。

使用场景： 在整个管道中维护 ID 值的运行总计或累计和，作为轻量级指标跟踪器。

$ java

List<Integer> transactions = List.of(100, 250, -50, 100);

// 计算从 0 开始的运行账户余额
List<Integer> balanceHistory = transactions.stream()
    .gather(Gatherers.scan(() -> 0, Integer::sum))
    .toList();

IO.println(balanceHistory); 
// 输出: [100, 350, 300, 400]

2.2.4 使用 `fold` 进行中间聚合

Gatherers.fold(initial, folder)：类似终结操作 reduce，但作为中间操作运行。它将所有元素折叠为单个结果，并只向下游发出该最终值。它严格按顺序执行。

使用场景： 让我们在管道中间将前五个故事 ID 组合成一个简洁的摘要字符串，以便在后续处理中保持井然有序。

$ java

List<Integer> storyIds = List.of(50028275, 50029123, 50019911, 50022292, 49997481, 49981264, 50023450);

// 将前 5 个 ID 折叠为单个逗号分隔的跟踪字符串
String summaryPayload = storyIds.stream()
    .limit(5)
    .gather(Gatherers.fold(
        () -> "Top Stories Snapshot: ",
        (accumulator, id) -> accumulator + "[" + id + "] "
    ))
    .findFirst()
    .orElse("");

IO.println(summaryPayload);
// 输出: Top Stories Snapshot: [50028275] [50029123] [50019911] [50022292] [49997481]

2.2.5 使用 `mapConcurrent` 实现有界并行

Gatherers.mapConcurrent(maxConcurrency, mapper)：该方法使用**虚拟线程（来自 Project Loom）**并发地对每个流元素应用映射函数，受最大并发限制约束，同时保持遇到顺序。

使用场景： 使用虚拟线程在有界并发限制下并发获取单个故事详情，同时保留获取顺序：https://hacker-news.firebaseio.com/v0/item/{id}.json。

$ java

import java.util.List;
import java.util.stream.Gatherers;

public class MapConcurrentDemo {

    public static void main(String[] args) {
        // 一小份 Hacker News 故事 ID 列表
        List<Integer> storyIds = List.of(39121, 39122, 39123, 39124, 39125);

        // 使用最多 3 个虚拟线程的并发上限并发获取标题
        List<String> titles = storyIds.stream()
            .gather(Gatherers.mapConcurrent(3, id -> fetchStoryTitle(id)))
            .toList();

        titles.forEach(System.out::println);
    }

    // 模拟 API 调用方法
    private static String fetchStoryTitle(int id) {
        try {
            // 模拟网络延迟（例如调用 Hacker News API）
            Thread.sleep(500); 
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "Story #" + id + " (Fetched on thread: " + Thread.currentThread().getName() + ")";
    }
}

有界并发（3）：即使流中包含数千个元素，它最多也只同时执行三个任务，从而防止下游服务或 API 因负载过高而不堪重负。
底层使用虚拟线程：它利用 Project Loom 的轻量级虚拟线程，确保阻塞式网络调用不会阻碍关键平台线程或操作系统线程的执行。
保持顺序：尽管任务并发执行，mapConcurrent 仍确保输出列表（titles）保持输入 storyIds 的精确顺序。

3. 进阶：从使用 Gatherers 到编写自己的 Gatherers

3.1 何时应编写自定义 Gatherer？

自定义 gatherer 可在以下场景中实现：

当需要内置操作（map、filter、collect）之外的额外控制时。
用于窗口化、扫描、去重等高级转换时。
用于性能优化且可复用的流处理逻辑时。

3.2 使用 `andThen()` 进行管道组合

Gatherers 通过 andThen(Gatherer) 方法支持组合，该方法将两个 gatherer 组合起来，其中第一个产生的元素可由第二个消费。这样可以像函数组合一样，通过组合更简单的 gatherer 来创建复杂的 gatherer。

从语义上说，source.gather(one).gather(two).gather(three).collect(...) 等价于 source.gather(one.andThen(two).andThen(three)).collect(...)。### 4. 深入底层：自定义 Gatherer 的架构与构成

4.1 `Gatherer<T, A, R>` 的构成

Gatherer 是一个包含三个参数的接口：

T：输入类型元素
A：Gatherer 的潜在可变状态类型
R：Gatherer 操作产生的输出元素类型

4.2 四个构建块（`initializer`、`integrator`、`combiner`、`finisher`）

Gatherer 由四个构建块组成：

[LOADING...]

initializer()：该构建块创建并返回一个新状态。
integrator()：该构建块将每个输入元素整合到状态中，并将结果推送到下游。
combiner()：该构建块将两个状态合并为一个。
finisher()：该构建块在输入被完全消费后执行一个可选的最终动作。

4.3 具体演练：构建自定义 Gatherer

设想你正在从 Hacker News API 获取持续不断的故事流。然而，由于频繁更新或置顶，重复的故事（或同一作者、同一域名下的故事）会反复出现。你的目标是实现一个自定义中间操作，基于唯一键提取器动态过滤元素，从而保留流的惰性求值并避免依赖外部集合。

步骤 1：定义数据模型

假设我们的 Hacker News 故事模型是一个简单的 Java record，如下所示：

$ java

public record HackerNewsStory(int id, String title, String author) {}

步骤 2：构建自定义 Gatherer

使用工厂方法构建自定义 Gatherer 时，你最多会与四个函数式构建块交互。integrator 在任何情况下都是必需的；不过，API 为其他函数提供了默认值，只有当你的 Gatherer 的特定行为需要时，才需要包含它们。

Initializer（Supplier）：该部分负责设置内部状态缓冲区，即一个用于记录我们已经见过的所有键的 HashSet。
Integrator（Integrator<A, R, T>）：该组件接收每个元素，并根据当前状态进行检查。如果它是新的，就将其推送到下游流中，并告诉我们可以继续。
Combiner（BinaryOperator）：如果流正在并行求值，该部分会将状态集合合并在一起。
Finisher（BiConsumer<A, Downstream>）：这是一个可选步骤。如果你希望在流结束时清理缓冲的多余元素，就可以使用它。

$ java

import java.util.HashSet;
import java.util.function.Function;
import java.util.stream.Gatherer;

public class HackerNewsGatherers {

    public static <T, K> Gatherer<T, ?, T> distinctBy(Function<? super T, ? extends K> keyExtractor) {
        return Gatherer.ofSequential(
            // 1. 初始化器：创建一个私有状态集合
            HashSet::new,
            
            // 2. 整合器：按顺序处理每个传入元素
            (state, element, downstream) -> {
                K key = keyExtractor.apply(element);
                if (state.add(key)) {
                    downstream.push(element);
                }
                return true; // 继续消费上游元素
            }
        );
    }
}

步骤 3：将自定义 distinctBy Gatherer 集成到流流水线中

现在，我们可以轻松地将自定义 distinctBy Gatherer 添加到流流水线中，以根据特定属性过滤故事，例如移除同一作者的重复提交。

$ java

import java.util.List;

public class Main {
    public static void main(String[] args) {
        List<HackerNewsStory> fetchedStories = List.of(
            new HackerNewsStory(1, "Java 27 Released", "mahi"),
            new HackerNewsStory(2, "Understanding Virtual Threads", "apj"),
            new HackerNewsStory(3, "Deep Dive into JEP 485", "mahi"), // 重复作者
            new HackerNewsStory(4, "Building Modern CLI Apps", "kate")
        );

        // 应用自定义 distinctBy Gatherer
        List<HackerNewsStory> uniqueAuthorStories = fetchedStories.stream()
            .gather(HackerNewsGatherers.distinctBy(HackerNewsStory::author))
            .toList();

        uniqueAuthorStories.forEach(story -> 
            IO.println("Author: " + story.author() + " -> Title: " + story.title())
        );
        
        // 输出：
        // Author: mahi -> Title: Java 27 Released
        // Author: apj -> Title: Understanding Virtual Threads
        // Author: kate -> Title: Building Modern CLI Apps
    }
}

5. 生产级 Gatherer：并行性、状态与陷阱

5.1 并行执行模式与 Combiner

问题：当你运行 parallelStream() 时，流运行时会会将数据分配给多个工作线程，分别处理每一部分，然后使用 combiner() 函数合并它们的中间状态对象（A）。
陷阱：如果你的自定义 Gatherer 没有 combiner()（或者默认状态无法合并），并行流可能无法按预期工作，或者可能静默降级为顺序求值。
实用建议：除非你的 Gatherer 被设计为在单线程中运行，或者不需要管理状态，否则请确保提供一个强健的 combiner（例如 (s1, s2) -> { s1.addAll(s2); return s1; }）。

5.2 贪婪式与短路式 Integrator

机制：可以把 Integrator 看作一个开关，它告诉你是继续还是停止。返回值要么是 true（继续消费），要么是 false（短路并停止）。
性能收益：对每个元素都检查这个开关会增加一点额外工作。如果你的自定义 Gatherer 会完整处理每个元素（例如 mapConcurrent 或 windowFixed），则可以使用 Integrator.ofGreedy(...)。
实用建议：使用 ofGreedy 相当于告诉 JVM 运行时你不需要检查短路，这有助于流水线运行得更快。

5.3 状态管理与线程安全

原则：由 initializer() supplier 创建的状态类型 A，会为处理流的每个线程或分段分别创建。
陷阱：绝不要在 Integrator 中使用外部的共享可变集合或静态变量。否则，当多个线程同时运行时，可能会引发隐藏的竞态条件。
实用建议：将状态仅保留在 lambda 体或工厂作用域内，以保持纯粹并确保线程安全。

5.4 Gatherer 与 Collector：快速比较

特性	流 Gatherer（`Gatherer<T, A, R>`）	流 Collector（`Collector<T, A, R>`）
流水线位置	中间操作（`.gather(...)`）	终端操作（`.collect(...)`）
处理语义	基于推送（向下游动态发出元素）	基于拉取/累积（将流折叠到容器中）
返回类型	返回新的 `Stream<R>`，允许继续链式调用	返回最终结果（例如 `List`、`Map`、`Long`）
可组合性	通过 `.andThen()` 高度可组合	独立的终端端点

6. 结论

自 Java 8 引入 Streams API 以来，许多 Java 开发者遇到了中间操作的架构限制。然而，终端操作可通过 Collector 扩展。随着 Stream Gatherers 的出现，这些限制得到解决，使开发者能够创建自定义中间操作。

本质上，Stream Gatherers 不仅提升了便利性，还实现了 Java Stream API 的功能愿景，使其从始至终完全可扩展。

参考资料

作者：Mahendra Rao B

我是一名来自印度的资深技术架构师，拥有超过 18 年的 Java、Spring、微服务和 AWS 经验。我热衷于构建可扩展的解决方案，并为 Spring、Arconia 和 OpenRewrite 等开源项目做出贡献。

分类导航

Java Stream Gatherers：从内置模式到自定义管道机制

Java Stream Gatherers：从内置模式到自定义管道机制

目录

1. 引言与背景

1.1 Streams 的演进

1.2 “Collector 问题”

1.3 什么是 Stream Gatherers？

2. 入门：内置 Gatherers 实战

2.1 理解流的基数

2.2 五个开箱即用的 Gatherers

2.2.1 使用 `windowFixed` 批量处理元素

2.2.2 使用 `windowSliding` 分析序列

2.2.3 使用 `scan` 累积值

2.2.4 使用 `fold` 进行中间聚合

2.2.5 使用 `mapConcurrent` 实现有界并行

3. 进阶：从使用 Gatherers 到编写自己的 Gatherers

3.1 何时应编写自定义 Gatherer？

3.2 使用 `andThen()` 进行管道组合

4.1 `Gatherer<T, A, R>` 的构成

4.2 四个构建块（`initializer`、`integrator`、`combiner`、`finisher`）

4.3 具体演练：构建自定义 Gatherer

5. 生产级 Gatherer：并行性、状态与陷阱

5.1 并行执行模式与 Combiner

5.2 贪婪式与短路式 Integrator

5.3 状态管理与线程安全

5.4 Gatherer 与 Collector：快速比较

6. 结论

参考资料

作者：Mahendra Rao B

分类导航

Java Stream Gatherers：从内置模式到自定义管道机制

目录

1. 引言与背景

1.1 Streams 的演进

1.2 “Collector 问题”

1.3 什么是 Stream Gatherers？

2. 入门：内置 Gatherers 实战

2.1 理解流的基数

2.2 五个开箱即用的 Gatherers

2.2.1 使用 windowFixed 批量处理元素

2.2.2 使用 windowSliding 分析序列

2.2.3 使用 scan 累积值

2.2.4 使用 fold 进行中间聚合

2.2.5 使用 mapConcurrent 实现有界并行

3. 进阶：从使用 Gatherers 到编写自己的 Gatherers

3.1 何时应编写自定义 Gatherer？

3.2 使用 andThen() 进行管道组合

4.1 Gatherer<T, A, R> 的构成

4.2 四个构建块（initializer、integrator、combiner、finisher）

4.3 具体演练：构建自定义 Gatherer

5. 生产级 Gatherer：并行性、状态与陷阱

5.1 并行执行模式与 Combiner

5.2 贪婪式与短路式 Integrator

5.3 状态管理与线程安全

5.4 Gatherer 与 Collector：快速比较

6. 结论

参考资料

作者：Mahendra Rao B

2.2.1 使用 `windowFixed` 批量处理元素

2.2.2 使用 `windowSliding` 分析序列

2.2.3 使用 `scan` 累积值

2.2.4 使用 `fold` 进行中间聚合

2.2.5 使用 `mapConcurrent` 实现有界并行

3.2 使用 `andThen()` 进行管道组合

4.1 `Gatherer<T, A, R>` 的构成

4.2 四个构建块（`initializer`、`integrator`、`combiner`、`finisher`）