JEP 461: Stream Gatherers (Preview) 总结

概述

JEP 461 已经完成并纳入 JDK 22。该提案旨在增强 Java 的 Stream API，支持自定义的中间操作，使得流管道能够以现有内置中间操作难以实现的方式转换数据。

核心内容

JEP 461 引入了一个名为 Gatherer<T, A, R> 的接口，作为 Stream API 的中间操作。该接口通过四个关键函数来处理输入元素（T），同时可选地改变中间状态（A），并将这些元素转换为不同类型的输出（R）：

1. 初始器（Initializer）：提供处理流元素的私有状态。
2. 整合器（Integrator）：负责整合输入流中的新元素，可以输出元素或提前终止处理。
3. 组合器（Combiner）：支持并行评估。
4. 结束器（Finisher）：在没有更多输入元素时调用，可能输出额外的元素。

动机

现有的 Stream API（自 Java 8 引入）虽然提供了丰富的中间和终端操作（如映射、过滤、归约和排序），但在处理复杂任务时缺乏灵活性。例如，基于字符串长度而非内容创建去重流并不直观，开发者需要编写额外的代码来实现此类功能，导致代码难以维护。

解决方案

JEP 461 引入了 Stream.gather(Gatherer)，这是一个新的中间操作，允许开发者使用自定义的 gatherer 处理流元素。Gatherer 可以进行一对一、一对多、多对一或多对多的转换，还可以跟踪之前看到的元素以影响后续元素的转换，支持短路操作（将无限流转换为有限流），并允许并行执行。

示例

1. 去重示例：

   var result = Stream.of("foo", "bar", "baz", "quux")
                      .map(DistinctByLength::new)
                      .distinct()
                      .map(DistinctByLength::str)
                      .toList();

   使用 gatherer 可以更简洁地实现类似功能。

2. 温度检测示例：

   List<List<Reading>> findSuspicious(Stream<Reading> source) {
       return source.gather(Gatherers.windowSliding(2))
                    .filter(window -> (window.size() == 2
                                       && isSuspicious(window.get(0),
                                                       window.get(1))))
                    .toList();
   }

   使用 windowSliding gatherer 可以更直观地检测异常温度变化。

内置 Gatherer

JEP 461 提供了多个内置的 gatherer，包括 fold、mapConcurrent、scan、windowFixed 和 windowSliding，这些 gatherer 可以直接使用或作为更复杂转换的基础。

示例

1. 固定窗口示例：

   List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5, 6, 7, 8)
                                       .gather(Gatherers.windowFixed(3))
                                       .toList();

   结果为 [[1, 2, 3], [4, 5, 6], [7, 8]]。

2. 折叠示例：

   Optional<String> numberString 
           = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
                   .gather(Gatherers.fold(() -> "", (string, number) -> string + number))
                   .findFirst();

   结果为 Optional[123456789]。

结论

JEP 461 通过引入 stream gatherer，增强了 Java Stream API 的功能，为开发者提供了更强大的工具来处理复杂的流处理任务，提升了代码的表达能力和灵活性。