Java 22 的流收集器，自定义流处理中间操作

Java 中的流终于可以自定义中间处理操作了。这是 Java 22 中新增的预览功能。

自从 Java 8 引入了 Stream 之后，Stream 在 Java 应用中的使用频率非常高。

Stream 处理时的 pipeline 由 3 个部分组成，分别是源、中间操作和终结操作：

• 源是 Stream 中元素的来源，可以通过 Stream.of 方法创建，或者从集合类得到。
• 中间操作对流中的元素进行处理，定义在 Stream 类中，包括 map、filter、flatMap、distinct 等。
• 终结操作会最终产生一个结果或者副作用，包括 collect、reduce、count、forEach 等

在使用 Stream 时，一个很大的痛点是中间操作不能自定义，只能用 Stream 类提供的那些。标准库中的 Stream 类也不会随意更改来添加更多的中间操作。这就导致在某些情况下，Stream 的使用很别扭。

举个例子，Stream 里面有中间操作 distinct，可以去除流中的重复元素，但是只能按照对象相等性来去重。在实际开发中，一个很常见的需求是按照对象的属性来去重。比如，Stream 中包含的是表示学生的 Student 对象。我们希望根据 Student 对象的 id 属性来去重。这个是 distinct 无法实现的。

Java 22 引入了Gatherer 接口，可以提供自定义的 Stream 中间操作。

static <T, A, R> Gatherer<T, A, R> of(
            Supplier<A> initializer,
            Integrator<A, T, R> integrator,
            BinaryOperator<A> combiner,
            BiConsumer<A, Downstream<? super R>> finisher) {
        return new Gatherers.GathererImpl<>(
                Objects.requireNonNull(initializer),
                Objects.requireNonNull(integrator),
                Objects.requireNonNull(combiner),
                Objects.requireNonNull(finisher)
        );
    }

一个 Gatherer 接口由4个方法组成， 分别是 initializer, integrator，combiner 和 finisher。只有 integrator 是必须的，其他都是可选的。

• initializer 创建一个新的可变状态，无状态的 Gatherer不需要提供。
• integrator 提供传递给下游的元素。
• combiner 在并行处理时使用，用来合并中间的状态，顺序处理时不需要提供。
• finisher 对最终的结果执行额外的处理。

使用 Gatherer 的 of 方法来创建 Gatherer，并由 Stream 的 gather 方法来使用。

下面展示一下如何用 Gatherer来实现基于对象属性的 distinct 操作。这个 Gatherer 实现采用顺序处理，因此不需要用到 combiner。

• State 表示中间状态，里面用一个 Map 来记录已经出现过的值。initializer 返回一个新的 State 对象。
• integrator 对于每个元素，调用提供的 extrator 函数，从元素对象中提取属性值。属性值作为 Map 的 key， value 则是元素本身。
• finisher 查看 State 中 Map 的全部值，并把这些值对象，通过 Downstream 的 push 方法添加进去，作为传递给下游的值。

/**
 * Distinct by an object's property, keep last duplicated element
 */
public class DistinctBy {

  /**
   * Create a new {@linkplain DistinctBy} gatherer
   *
   * @param extractor Extract the property from an element
   * @param <T>       Type of the element
   * @param <R>       Type of the element's property
   * @return A new {@linkplain DistinctBy} gatherer
   */
  static <T, R> Gatherer<T, ?, T> of(Function<? super T, ? extends R> extractor) {
    class State {

      final Map<R, T> seen = new HashMap<>();
    }

    return Gatherer.ofSequential(State::new, Integrator.ofGreedy(
            ((state, element, _) -> {
              state.seen.put(extractor.apply(element), element);
              return true;
            })),
        (state, downstream) -> state.seen.values().forEach(downstream::push)
    );
  }
}

下面的代码展示了如何使用这个 Gatherer。调用 Stream 的 gather 方法，并传入 Gatherer 对象，整体的代码清晰易懂。

public class CustomGatherers {

  record Student(String id, String name) {

  }

  void distinctBy() {
    var result = Stream.of(
            new Student("001", "Alex"),
            new Student("002", "Bob"),
            new Student("001", "Alex")
        ).gather(DistinctBy.of(Student::id))
        .toList();
    // [Student[id=001, name=Alex], Student[id=002, name=Bob]]
    System.out.println(result);
  }

  public static void main(String[] args) {
    new CustomGatherers().distinctBy();
  }
}

与 Gatherer 接口同时发布的还有几个内置的 Gatherer 实现，在 Gatherers 这个类里面。

这些内置的实现包括：

• windowFixed 把元素收集到指定大小的窗口中。
• windowSliding 把元素收集到指定大小的滑动窗口中。
• fold 类似 reduce，产生一个值给下游。
• scan 从一个初始值开始，后续的每个值根据当前值和流中的输入元素，计算而来。所有这些值都会被提供给下游。
• mapConcurrent 执行并发的 map 操作，可以指定同时运行的最大任务数量

流收集器目前还是预览功能，我们大概率会在 Java的 下一个 LTS Java 25 中使用它。