深度解析java.lang.util.stream

接 java.lang.util.stream包解析 一文，深入剖析 Stream 的实现。

一、Stream API 的接口抽象设计

Stream API 的核心设计思想是解耦数据源与操作，并通过流水线（Pipeline）和惰性求值（Lazy Evaluation）实现高效处理。其接口抽象分为以下层次：

1. 核心接口分层

BaseStream<T, S extends BaseStream<T, S>>  // 基础流接口（定义关闭、并行等方法）
   └── Stream<T>                           // 具体流接口（定义 map、filter 等操作）
   └── IntStream/LongStream/DoubleStream   // 特化流（处理基本类型）

• BaseStream: 定义了流的通用行为（如parallel()、sequential()、close()）。
• Stream<T>: 针对对象类型的流，提供丰富的中间/终端操作（map, filter, collect等）。
• 特化流（如IntStream）: 避免装箱开销，优化性能。

2. 操作类型抽象

Stream 操作通过函数式接口定义行为，例如：

• Predicate<T> (过滤条件)
• Function<T, R> (映射逻辑)
• Consumer<T> (消费元素)
• Supplier<T> (生成元素)
• Collector<T, A, R> (自定义收集策略)

二、Stream 的实现机制

Stream 的底层实现基于流水线（Pipeline）模型，由多个阶段（Stage）构成。每个阶段代表一个中间操作或终端操作。

1. 流水线结构

一个典型的 Stream 流水线结构如下：

List<String> result = list.stream()          // Source (数据源)
       .filter(s -> s.length() > 3)          // Stage 1 (过滤)
       .map(String::toUpperCase)             // Stage 2 (转换)
       .collect(Collectors.toList());        // Terminal (终端操作)

• 数据源（Source）：可以是集合、数组、生成器等。
• 中间操作（Intermediate Stages）：每个操作生成一个新的流水线阶段。
• 终端操作（Terminal Stage）：触发整个流水线的执行。

2. 惰性求值与流水线链接

中间操作是惰性的，它们仅记录操作逻辑，并不立即执行。只有在终端操作被调用时，整个流水线才会开始处理数据。

实现关键：

• AbstractPipeline 类：核心实现类，维护流水线各阶段的状态（如源数据、操作链）。
• Sink 接口：数据处理的“水槽”，每个操作对应一个 Sink，多个 Sink 通过链表连接，形成处理链。

示例：filter 和 map 的流水线链接

// 伪代码展示 Sink 链的构建
Sink<String> sink1 = new FilterSink<>(s -> s.length() > 3);
Sink<String> sink2 = new MapSink<>(String::toUpperCase);

sink1.downstream = sink2;  // 将两个 Sink 链接

当数据流过时，会依次经过 FilterSink 和 MapSink。

3. 并行处理机制

Stream 的并行能力基于 Fork/Join 框架 实现：

• Spliterator 接口：负责分割数据源（如集合），将数据分成多个块供并行处理。
• ForkJoinPool 类：实际执行并行任务的线程池。

关键步骤：

1. 数据源通过 Spliterator 分割为多个子任务。
2. 每个子任务独立处理自己的数据块。
3. 结果合并（如使用 Collectors.toList()）。

三、自定义 Stream 操作的实现示例

为了更深入理解，我们尝试实现一个简单的自定义中间操作：takeWhile（取元素直到条件不满足）。

1. 定义操作逻辑

public static <T> Stream<T> takeWhile(Stream<T> source, Predicate<T> predicate) {
    return StreamSupport.stream(
        new TakeWhileSpliterator<>(source.spliterator(), predicate),
        source.isParallel()
    ).onClose(source::close);
}

2. 实现 Spliterator

static class TakeWhileSpliterator<T> implements Spliterator<T> {
    private final Spliterator<T> source;
    private final Predicate<T> predicate;
    private boolean conditionMet = true;

    TakeWhileSpliterator(Spliterator<T> source, Predicate<T> predicate) {
        this.source = source;
        this.predicate = predicate;
    }

    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        if (!conditionMet) return false;
        return source.tryAdvance(item -> {
            if (predicate.test(item)) {
                action.accept(item);
            } else {
                conditionMet = false;
            }
        });
    }

    // 其他方法（如 estimateSize, characteristics）省略
}

3. 使用自定义操作

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6);
List<Integer> result = takeWhile(numbers.stream(), n -> n < 4)
                             .collect(Collectors.toList());
// 结果: [1, 2, 3]

四、设计总结

Java Stream API 的底层实现通过以下关键设计实现高效和灵活：

• 解耦数据源与操作：通过 Spliterator 抽象数据源，通过 Sink 链式处理操作。
• 惰性求值：仅在终端操作触发时执行计算，优化资源使用。
• 并行透明：通过 ForkJoinPool 和 Spliterator 实现并行处理，对开发者透明。