Java 容器之 Stream

简介

Stream 是 Java 8 引入的核心特性之一，位于 java.util.stream 包中。它提供了一种高效且可声明的方式来处理数据序列（集合、数组、文件行等），支持链式操作、函数式编程风格和并行处理。

Stream 不是数据结构，它不会存储数据，而是对 Collection 或数组等数据源进行惰性求值（Lazy Evaluation）的操作管道。Stream 操作不会修改数据源，而是返回一个新的 Stream 或结果。

在 Java 8 中，Collection 新增了两个流方法：stream() 和 parallelStream()。

特性

声明式编程

Stream API 让代码更加简洁、可读，开发者只需声明"做什么"而非"怎么做"：

// 传统方式：找出所有偶数并求和
int sum = 0;
for (int n : numbers) {
    if (n % 2 == 0) {
        sum += n;
    }
}

// Stream 方式
int sum = numbers.stream()
    .filter(n -> n % 2 == 0)
    .mapToInt(Integer::intValue)
    .sum();

链式操作

Stream 支持方法链（Fluent API），多个操作可以串联成一条管道：

List<String> result = users.stream()
    .filter(u -> u.getAge() > 18)
    .sorted(Comparator.comparing(User::getName))
    .map(User::getName)
    .collect(Collectors.toList());

惰性求值

中间操作不会立即执行，只有在终结操作触发时才会真正计算。这使得 Stream 能够进行短路优化（如 findFirst、limit），避免不必要的计算。

可并行化

通过 parallelStream() 可以轻松实现并行处理，底层基于 Fork/Join 框架：

long count = list.parallelStream()
    .filter(n -> n > 0)
    .count();

原理

Stream 操作分类

Stream 操作分为两大类：

graph TB
    A[Stream 操作] --> B[中间操作 Intermediate]
    A --> C[终结操作 Terminal]
    B --> B1[无状态操作 Stateless]
    B --> B2[有状态操作 Stateful]
    B1 --> B1a["filter / map / flatMap"]
    B2 --> B2a["sorted / distinct / limit"]
    C --> C1[短路操作 Short-circuiting]
    C --> C2[非短路操作]
    C1 --> C1a["findFirst / findAny / anyMatch"]
    C2 --> C2a["forEach / collect / reduce / count"]

• 中间操作：返回一个新的 Stream，支持链式调用。分为无状态（filter、map）和有状态（sorted、distinct）操作。
• 终结操作：触发管道执行并产生结果或副作用，如 collect、forEach、reduce。

Stream 内部架构

graph TB
    A[数据源 Collection/Array] --> B["stream() / parallelStream()"]
    B --> C[Head - 数据源 Stage]
    C --> D[中间操作 Stage 链表]
    D --> D1[StatelessOp - filter/map]
    D --> D2[StatefulOp - sorted/distinct]
    D1 --> E[终结操作 Stage]
    D2 --> E
    E --> F[触发执行 - Sink 链递归调用]
    F --> G[结果]

ReferencePipeline 是核心实现类，定义了 Head、StatelessOp、StatefulOp 三个内部类。整个 Stream 管道通过 Sink 接口协议串联，终结操作触发时从最后一个 Stage 开始递归生成 Sink 链。

典型应用场景

场景一：数据过滤与转换

从用户列表中筛选活跃用户并提取邮箱：

List<String> emails = users.stream()
    .filter(User::isActive)
    .map(User::getEmail)
    .collect(Collectors.toList());

场景二：分组与聚合统计

按部门分组并统计各部门人数和平均薪资：

Map<String, DoubleSummaryStatistics> stats = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::getDepartment,
        Collectors.summarizingDouble(Employee::getSalary)
    ));

场景三：日志分析与统计

处理大量日志数据，统计各状态码的请求数：

Map<Integer, Long> statusCount = logEntries.stream()
    .filter(entry -> entry.getTimestamp() > yesterday)
    .collect(Collectors.groupingBy(
        LogEntry::getStatusCode,
        Collectors.counting()
    ));

场景四：文件内容处理

读取文件并统计词频：

Map<String, Long> wordCount = Files.lines(Path.of("data.txt"))
    .flatMap(line -> Arrays.stream(line.split("\\s+")))
    .map(String::toLowerCase)
    .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));

最佳实践

1. 避免嵌套 Stream：过度嵌套会使代码难以阅读和调试，考虑拆分为多个方法。
2. 注意并行 Stream 的开销：对于小数据量，并行 Stream 可能比串行更慢（线程创建和协调开销）。数据量大于 10000 且操作耗时才考虑并行。
3. 优先使用方法引用：User::getName 比 u -> u.getName() 更简洁。
4. 避免副作用：不要在 filter/map 中修改外部状态，保持函数无副作用。
5. 使用 Optional 处理空值：findFirst()、max() 等返回 Optional，避免 NPE。
6. 终结操作只执行一次：Stream 是一次性使用的，终结操作后不能再复用。
7. 合理选择收集器：toList()、toMap()、groupingBy() 各有适用场景，避免手动循环收集。

常见问题

Q1：Stream 和 Iterator 有什么区别？

特性	Stream	Iterator
遍历方式	内部迭代（声明式）	外部迭代（命令式）
惰性求值	支持	不支持
并行处理	原生支持 parallelStream	不支持
可复用性	一次性使用	可多次遍历
函数式操作	支持 filter/map/reduce	不支持

Q2：parallelStream 一定比 stream 快吗？

不一定。并行 Stream 适用于数据量大且每个元素处理耗时较长的场景。对于小数据集，线程切换和协调开销可能抵消并行收益。此外，某些有状态操作（如 sorted、distinct）在并行模式下性能可能更差。

Q3：Stream 中 map 和 flatMap 有什么区别？

• map：一对一映射，每个元素转换为一个新元素。
• flatMap：一对多映射，每个元素转换为一个 Stream，然后将所有 Stream 扁平化合并。

// map: List<String> -> List<String>
list.stream().map(String::toUpperCase);

// flatMap: List<List<Integer>> -> List<Integer>
nestedList.stream().flatMap(Collection::stream);

参考资料

• Java 编程思想（第 4 版）
• 《Java 8 实战》
• java8-tutorial - Stream
• Java Stream API 官方文档

---

title: Java 容器之 Stream
date: 2020-12-05 18:30:22
order: 06
categories:

• Java
• JavaCore
• 容器
  tags:
• Java
• JavaCore
• 容器
  permalink: /pages/c7769e0d/

---

Java 容器之 Stream

Stream 简介

在 Java8 中，Collection 新增了两个流方法，分别是 stream() 和 parallelStream()。

Stream 相当于高级版的 Iterator，他可以通过 Lambda 表达式对集合进行各种非常便利、高效的聚合操作（Aggregate Operation），或者大批量数据操作 (Bulk Data Operation)。

Stream 操作分类

官方将 Stream 中的操作分为两大类：中间操作（Intermediate operations）和终结操作（Terminal operations）。

中间操作又可以分为无状态（Stateless）与有状态（Stateful）操作，前者是指元素的处理不受之前元素的影响，后者是指该操作只有拿到所有元素之后才能继续下去。

终结操作又可以分为短路（Short-circuiting）与非短路（Unshort-circuiting）操作，前者是指遇到某些符合条件的元素就可以得到最终结果，后者是指必须处理完所有元素才能得到最终结果。

Stream 源码实现

BaseStream 和 Stream 是最顶层的接口类。BaseStream 主要定义了流的基本接口方法，例如，spliterator、isParallel 等；Stream 则定义了一些流的常用操作方法，例如，map、filter 等。

Sink 接口是定义每个 Stream 操作之间关系的协议，他包含 begin()、end()、cancellationRequested()、accpt() 四个方法。ReferencePipeline 最终会将整个 Stream 流操作组装成一个调用链，而这条调用链上的各个 Stream 操作的上下关系就是通过 Sink 接口协议来定义实现的。

ReferencePipeline 是一个结构类，他通过定义内部类组装了各种操作流。他定义了 Head、StatelessOp、StatefulOp 三个内部类，实现了 BaseStream 与 Stream 的接口方法。Head 类主要用来定义数据源操作，在初次调用 names.stream() 方法时，会加载 Head 对象，此时为加载数据源操作；接着加载的是中间操作，分别为无状态中间操作 StatelessOp 对象和有状态操作 StatefulOp 对象，此时的 Stage 并没有执行，而是通过 AbstractPipeline 生成了一个中间操作 Stage 链表；当我们调用终结操作时，会生成一个最终的 Stage，通过这个 Stage 触发之前的中间操作，从最后一个 Stage 开始，递归产生一个 Sink 链。

Stream 并行处理

Stream 处理数据的方式有两种，串行处理和并行处理。

4. 参考资料

• Java 编程思想（第 4 版）