Lesson 39 · Java 高级特性

Stream API 深度使用：map、flatMap、reduce、Collector

中级·🔥 极高·#Java·#函数式

第 1 站

Stream 的本质：不是数据结构，是计算管道

"Stream 和集合有什么区别？它到底存不存数据？"

Stream 是 JDK 8 引入的函数式数据处理抽象。它不存储数据，只是对数据源（集合、数组、I/O 通道）执行一系列操作。你可以把它想象成一条流水线：原料进去，成品出来。

Stream 有五大核心特性：

惰性求值：中间操作不会立即执行，只有终端操作触发时才计算
内部迭代：由框架控制迭代，开发者声明"做什么"而非"怎么做"
可消费性：一个 Stream 只能被消费一次
链式组合：中间操作返回 Stream，可以无限链式调用
支持并行：parallelStream() 自动利用多核

创建 Stream 的六种常见方式：

StreamCreation.java · 六种创建方式

// 1. 从集合创建
List<String> list = Arrays.asList("a", "b", "c");
Stream<String> s1 = list.stream();
Stream<String> s2 = list.parallelStream();

// 2. 从数组创建
String[] arr = {"a", "b", "c"};
Stream<String> s3 = Arrays.stream(arr);

// 3. Stream.of()
Stream<Integer> s4 = Stream.of(1, 2, 3);

// 4. 从文件创建（JDK 8+）
Stream<String> lines = Files.lines(Paths.get("data.txt"));

// 5. 无限流 — iterate（必须配合 limit）
Stream<Integer> even = Stream.iterate(0, n -> n + 2);

// 6. 无限流 — generate
Stream<Double> randoms = Stream.generate(Math::random);

面试考什么？

面试官会问"无限流怎么安全使用？"——答案是必须配合 limit() 或 takeWhile()（JDK 9+）截断，否则终端操作会无限执行。

第 2 站

中间操作：map、flatMap、filter、sorted、distinct

中间操作的特点是返回 Stream 本身，可以链式组合。它们在终端操作触发前不会执行任何计算。

图 1 Stream 管道——惰性求值，从终端操作反向拉取

map vs flatMap 是面试高频题：map 是 1 对 1 转换，flatMap 是 1 对 N 转换后扁平化合并。

MapVsFlatMap.java · 核心区别

List<List<Integer>> nested = Arrays.asList(
    Arrays.asList(1, 2),
    Arrays.asList(3, 4, 5));

// map: 结果仍然是嵌套 List<List<Integer>>
List<List<Integer>> mapped = nested.stream()
    .map(inner -> inner.stream()
        .map(x -> x * 2)
        .collect(Collectors.toList()))
    .collect(Collectors.toList());
// [[2, 4], [6, 8, 10]]

// flatMap: 扁平化为一维 List<Integer>
List<Integer> flat = nested.stream()
    .flatMap(inner -> inner.stream().map(x -> x * 2))
    .collect(Collectors.toList());
// [2, 4, 6, 8, 10]

记忆口诀：

map: 1 → 1 | flatMap: 1 → N → 扁平化

filter 可以链式组合多个条件，效果等同于 AND。若需要 OR，在单个 filter 中用 || 连接：

FilterExamples.java · 多条件过滤

List<String> words = Arrays.asList("java", "Stream", "API", "JAVA8");

// 多 filter 链 = AND
List<String> result = words.stream()
    .filter(w -> w.length() > 3)
    .filter(w -> w.toUpperCase().startsWith("J"))
    .map(String::toUpperCase)
    .collect(Collectors.toList());
// ["JAVA", "JAVA8"]

// 非空过滤
List<String> nonNull = list.stream()
    .filter(Objects::nonNull)
    .collect(Collectors.toList());

性能提示

把 filter 放在 map 前面。先过滤再转换，减少不必要的对象创建和函数调用开销。

第 3 站

终端操作：触发管道执行的"最后一击"

终端操作是 Stream 管道的终点。没有终端操作，所有中间操作只是"声明"，不会执行任何计算。

终端操作	返回类型	说明
`collect(Collector)`	集合/Map/String	最常用，将结果收集到容器中
`reduce(identity, fn)`	T / Optional	归约为单个值
`forEach(Consumer)`	void	遍历消费每个元素
`count()`	long	元素个数
`anyMatch/allMatch/noneMatch`	boolean	短路匹配操作
`findFirst/findAny`	Optional	查找操作
`min/max`	Optional	最值操作

ReduceExample.java · 三种 reduce 用法

List<Integer> nums = Arrays.asList(1, 2, 3, 4, 5);

// 方式 1：无初始值（返回 Optional）
Optional<Integer> sum1 = nums.stream()
    .reduce((a, b) -> a + b);   // Optional[15]

// 方式 2：有初始值
int sum2 = nums.stream()
    .reduce(0, Integer::sum);   // 15

// 方式 3：带 combiner（并行流用）
int sum3 = nums.parallelStream()
    .reduce(0,
        Integer::sum,           // accumulator
        Integer::sum);          // combiner

forEach vs forEachOrdered：在并行流中，forEach 不保证处理顺序，forEachOrdered 保证按遇到顺序处理。

面试怎么答？

"reduce 在并行流中使用时，accumulator 和 combiner 必须满足结合律和交换律，否则结果不确定。求和、求积天然满足，但字符串拼接用 StringBuilder 就不满足交换律，这时候应该用 collect。"

第 4 站

Collector 框架：groupingBy、partitioningBy、自定义收集器

Collectors 工具类提供了丰富的收集器。面试中最常考的是 groupingBy + downstream 的嵌套用法。

图 2 Collector 三要素与高频收集器速查

CollectorShowcase.java · groupingBy 嵌套用法

// 按部门分组，计算各部门平均薪资
Map<String, Double> avgByDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::getDept,
        Collectors.averagingDouble(Employee::getSalary)
    ));

// 按部门分组 → 每组薪资最高员工
Map<String, Employee> topPerDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::getDept,
        Collectors.collectingAndThen(
            Collectors.reducing((a, b) ->
                a.getSalary() > b.getSalary() ? a : b),
            Optional::get)
    ));

// 分区：薪资 > 10000 与 <= 10000
Map<Boolean, List<Employee>> partitioned = employees.stream()
    .collect(Collectors.partitioningBy(
        e -> e.getSalary() > 10000));

// 字符串拼接
String names = employees.stream()
    .map(Employee::getName)
    .collect(Collectors.joining(", ", "[", "]"));
// [Alice, Bob, Charlie]

自定义 Collector 四步走

Collector.of(supplier, accumulator, combiner, finisher)——当内置收集器无法满足需求时，用这四个函数构建自己的收集器。串行流时 combiner 不会被调用，但并行流中必不可少。

第 5 站

并行流：parallelStream 的机遇与陷阱

"把 stream() 换成 parallelStream() 就能变快吗？"

答案是：大多数情况下不会变快，反而可能更慢甚至出错。parallelStream 底层基于 ForkJoinPool 的 commonPool，线程数默认为 CPU 核心数 - 1。

并行流适用的场景：

数据量大（通常 > 10,000 个元素）
计算密集（CPU 密集型操作）
Lambda 无副作用（不修改外部共享状态）
操作无状态（不依赖 sorted/distinct/limit）

并行流的常见陷阱：

ParallelPitfalls.java · 线程安全陷阱

// ❌ 陷阱 1：并行中修改共享集合（线程不安全）
List<String> result = new ArrayList<>();
employees.parallelStream()
    .filter(e -> e.getSalary() > 10000)
    .forEach(e -> result.add(e.getName())); // 并发修改！

// ✅ 正确：用 collect 代替 forEach + 共享集合
List<String> safe = employees.parallelStream()
    .filter(e -> e.getSalary() > 10000)
    .map(Employee::getName)
    .collect(Collectors.toList());

// ❌ 陷阱 2：有状态操作 sorted + parallel
list.parallelStream()
    .sorted()        // 需要全局排序，并行优势被抵消
    .limit(10)       // 需要全局排序后才能截取
    .collect(Collectors.toList());

// ❌ 陷阱 3：装箱开销
int sum = list.stream()
    .map(String::length)        // 返回 Stream<Integer>，装箱！
    .reduce(0, Integer::sum);

// ✅ 用 IntStream 避免装箱
int sum = list.stream()
    .mapToInt(String::length)   // 返回 IntStream，无装箱
    .sum();

面试答题模板

"parallelStream 基于 ForkJoinPool 的 commonPool，适合数据量大、计算密集、无副作用的场景。但要注意三点：不在 Lambda 里修改共享集合、避免 sorted/limit 等有状态操作、用 IntStream/LongStream 代替装箱流。数据量小于 1 万时，串行通常更快。"

第 6 站

实战场景：员工数据分析全流程

把前面学到的所有操作组合起来，解决一个真实的业务问题：

EmployeeAnalysis.java · 完整实战示例

// 场景 1：销售部门 30 岁以上员工，按薪资降序
List<String> names = employees.stream()
    .filter(e -> "Sales".equals(e.getDept()))
    .filter(e -> e.getAge() > 30)
    .sorted(Comparator.comparingDouble(
        Employee::getSalary).reversed())
    .map(Employee::getName)
    .collect(Collectors.toList());

// 场景 2：统计各部门人数和平均薪资
Map<String, Long> countByDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::getDept,
        Collectors.counting()));

// 场景 3：薪资最高的 Top 3
List<Employee> top3 = employees.stream()
    .sorted(Comparator.comparingDouble(
        Employee::getSalary).reversed())
    .limit(3)
    .collect(Collectors.toList());

// 场景 4：词频统计
Map<String, Long> wordCount = Arrays.stream(text.split(" "))
    .collect(Collectors.groupingBy(
        Function.identity(),
        Collectors.counting()));

Stream 编程思维模型：先确定数据源 → 链式加中间操作（filter/map/sorted） → 选终端操作（collect/reduce/forEach）。

Stream 管道设计公式：

数据源.stream().filter(...).map(...).sorted(...).collect(...)

中间操作可自由组合顺序，无需关心执行时机——惰性求值会保证效率。

性能清单

(1) filter 放 map 前面；(2) 用 IntStream/LongStream 避免装箱；(3) 大数据量考虑 parallelStream；(4) 用 peek() 调试中间结果；(5) Stream 链控制在 3~5 步以内，太长考虑拆分。

第 7 站

总结：Stream API 全景回顾

全文核心要点回顾

Stream 本质：不是数据结构，是对数据源执行函数式操作的计算管道
惰性求值：中间操作不执行，终端操作触发时才逐元素计算
map vs flatMap：map 是 1→1 转换，flatMap 是 1→N 后扁平化
collect：最常用的终端操作，配合 Collectors 工具类可以分组、分区、聚合
reduce：将流归约为单个值，并行流中需满足结合律
parallelStream：基于 ForkJoinPool，适合大数据量 + 计算密集 + 无副作用
性能优化：filter 前置、避免装箱、不滥用并行

面试 30 秒总结

"Stream 是 JDK 8 的函数式数据处理 API，核心是惰性求值——中间操作只声明，终端操作才执行。map 做一对一转换，flatMap 做一对多再扁平化。collect 配合 groupingBy/partitioningBy 可以实现复杂聚合。reduce 用于归约，并行流使用时需满足结合律。parallelStream 基于 ForkJoinPool，适合大数据量计算密集场景，但要注意线程安全和装箱开销。"