07.2 — Stream 流水线
定位: 从"怎么做"到"做什么" — 声明式数据处理的核心机制 面试高频度: ⭐⭐⭐⭐ 考查方式: 中间操作 vs 终端操作、惰性求值、并行流原理
一、这是什么？为什么需要它？
从外部迭代到内部迭代
java
// 传统：外部迭代（告诉计算机怎么做）
List<String> result = new ArrayList<>();
for (String s : list) {
if (s.startsWith("A")) {
result.add(s.toUpperCase());
}
}
// Stream：内部迭代（告诉计算机要什么）
List<String> result = list.stream()
.filter(s -> s.startsWith("A"))
.map(String::toUpperCase)
.collect(Collectors.toList());为什么需要 Stream？
1. 声明式 — 聚焦"要什么"，而非"怎么做"
2. 可组合 — filter → map → collect 像流水线
3. 惰性求值 — 只在最后一次性计算
4. 并行友好 — parallelStream() 一键并行
5. 避免副作用 — 不修改源数据二、原理拆解
2.1 Stream 的三阶段模型
┌─────────────────────────────────────────────────────────────┐
│ Stream 操作三阶段 │
│ │
│ 数据源 ──→ 中间操作(1..N) ──→ 终端操作 │
│ (集合/ filter collect │
│ 数组/ map forEach │
│ 文件) sorted reduce │
│ flatMap count │
│ distinct ... │
│ limit │
│ skip │
│ │
│ 中间操作：惰性（lazy），构建操作流水线，不执行 │
│ 终端操作：触发执行（eager），消费 Stream，完后 Stream 关闭 │
│ │
│ 关键：没有终端操作，中间操作不会执行！ │
└─────────────────────────────────────────────────────────────┘2.2 惰性求值
java
Stream<String> stream = list.stream()
.filter(s -> {
System.out.println("filter: " + s);
return s.length() > 3;
})
.map(s -> {
System.out.println("map: " + s);
return s.toUpperCase();
});
// 到目前为止，没有输出！—— 中间操作是惰性的
List<String> result = stream.collect(Collectors.toList());
// 终端操作触发 → 现在才执行！惰性求值的意义：
不需要将所有中间结果保存在内存中。
例：从 1000 万条数据中找到前 10 个匹配项
list.stream()
.filter(expensivePredicate)
.limit(10)
.collect(toList());
如果没有惰性求值：
filter → 计算 1000 万次 → 生成 1000 万中间结果
limit → 取前 10 个 → 剩余 9999990 个丢弃！
有惰性求值：
filter 和 limit 融合 → filter 找到 10 个就停止
→ 只计算约 10+ 次
⚡ 短路的威力：limit、findFirst、anyMatch 等短路操作2.3 中间操作分类
无状态操作（每个元素独立处理）：
filter() — 过滤
map() — 转换
flatMap() — 展平
peek() — 查看（调试用）
forEach() — 遍历（终端操作）
有状态操作（需要记录状态）：
distinct() — 去重（需要保存已见过的元素）
sorted() — 排序（需要处理全部元素）
limit() — 限制（需要计数）
skip() — 跳过（需要计数）2.4 并行流原理
java
// 一键并行
list.parallelStream()
.filter(...)
.map(...)
.collect(toList());
// 等价于
list.stream()
.parallel()
.filter(...)
.map(...)
.collect(toList());并行流底层：ForkJoinPool.commonPool()
工作原理：
① 分割：将数据源分成多个段（Spliterator）
② 处理：每个段分配给一个线程并行处理
③ 合并：将各段结果合并
适合并行：
- 数据量大（>10000）
- 每个元素处理耗时长
- 无状态操作（filter/map）
不适合并行：
- 数据量小（并行开销 > 收益）
- 有状态操作（sorted/distinct 有合并成本）
- 顺序依赖（limit/findFirst 需要串行语义）
- 阻塞操作（I/O — 会占用公共池线程）三、图解全景
┌─────────────────────────────────────────────────────────────┐
│ Stream 流水线执行示意图 │
│ │
│ 数据: ["Alice", "Bob", "Amy", "David", "Anna"] │
│ │
│ 操作链: filter(name -> name.startsWith("A")) │
│ .map(String::toUpperCase) │
│ .limit(2) │
│ .collect(toList()) │
│ │
│ 执行（垂直处理——每个元素走完全程）： │
│ │
│ "Alice" → filter(A) ✅ → map → "ALICE" → limit(1) → 收集 │
│ "Bob" → filter(B) ❌ → (跳过) │
│ "Amy" → filter(A) ✅ → map → "AMY" → limit(2) → 收集 │
│ "David" → filter(D) ❌ → (跳过) │
│ "Anna" → filter — limit 已满 → 跳过(不再处理) │
│ │
│ 结果: ["ALICE", "AMY"] │
│ │
│ ⚡ 注意：不是先全部 filter，再全部 map！ │
│ 而是一个元素走完整个流水线再处理下一个 │
│ 这使 limit 能够提前终止 │
└─────────────────────────────────────────────────────────────┘四、实战验证
4.1 惰性求值验证
java
List<String> list = Arrays.asList("Alice", "Bob", "Amy", "David");
long count = list.stream()
.filter(s -> {
System.out.println("filter: " + s);
return s.length() > 3;
})
.count(); // 终端操作
// 输出（按元素垂直执行）：
// filter: Alice
// filter: Bob
// filter: Amy
// filter: David
// 结果: count = 24.2 并行流性能
java
long count = LongStream.rangeClosed(1, 10_000_000)
.filter(n -> n % 2 == 0)
.count();
// 串行: ~50ms
// 并行: ~15ms
// 但如果是轻量操作且数据量小：
IntStream.range(1, 1000)
.parallel()
.map(x -> x * 2) // 简单操作
.sum();
// 并行反而更慢（线程创建/调度开销）五、面试视角
| 追问 | 答案要点 |
|---|---|
| Stream 中间操作和终端操作的区别？ | 中间操作惰性构建流水线；终端操作触发执行并关闭 Stream |
| 惰性求值有什么好处？ | 短路优化（limit）、减少中间对象、流水线融合优化 |
| parallelStream 一定比 stream 快吗？ | 不一定。数据量大+元素处理耗时+无状态操作才适合 |
| parallelStream 使用的线程池？ | ForkJoinPool.commonPool()，默认线程数=CPU核心数-1 |
| Stream 能复用吗？ | 不能。终端操作后 Stream 已关闭，需重新创建 |
| forEach 和 for-each 循环哪个快？ | 小数据量 for-each 快；大数据量+复杂操作 Stream 快 |
📚 相关链接
- **Lambda与函数式接口** — Lambda 语法基础
- **Collectors收集器** — 收集数据的方法
- **集合索引** — 集合作为 Stream 的数据源
- ← 返回 **Lambda索引**