Stream 与函数式处理

流水线模型：源 → 中间 → 终端01

Stream 本身不存储数据，它描述一条处理流水线。中间操作（map、filter、sorted）返回新的 Stream 且惰性；终端操作（collect、forEach、count）触发实际计算，之后 Stream 不可复用。

List<String> result = names.stream()   // 源
        .filter(n -> !n.isBlank())       // 中间：惰性
        .map(String::toUpperCase)        // 中间：惰性
        .sorted()                        // 中间：有状态，仍惰性
        .limit(3)                        // 中间：短路
        .toList();                       // 终端：触发执行

// Stream 只能消费一次
Stream<String> s = Stream.of("a", "b");
s.forEach(System.out::println);
s.forEach(System.out::println); // IllegalStateException: stream has already been operated upon

惰性与短路是 Stream 的性能基础：limit(3) 遇到前 3 个元素后就不再向上游要数据；anyMatch 找到第一个匹配就返回。对无限流来说，短路操作是唯一安全的终端操作。

Stream.iterate(1, n -> n * 2)
      .limit(10)
      .forEach(n -> System.out.print(n + " "));
// 1 2 4 8 16 32 64 128 256 512

Optional<Integer> first = Stream.iterate(1, n -> n + 1)
      .filter(n -> n % 7 == 0)
      .findFirst();
System.out.println(first); // Optional[7]

常用操作地图02

record Order(String user, int amount) {}
List<Order> orders = List.of(
    new Order("alice", 120),
    new Order("bob", 80),
    new Order("alice", 200));

// 按用户分组求和
Map<String, Integer> totals = orders.stream()
    .collect(Collectors.groupingBy(Order::user, Collectors.summingInt(Order::amount)));
System.out.println(totals); // {bob=80, alice=320}

// 每个用户取金额最大的订单
Map<String, Order> top = orders.stream()
    .collect(Collectors.toMap(Order::user, o -> o,
        BinaryOperator.maxBy(Comparator.comparingInt(Order::amount))));
System.out.println(top);

// 基本类型流避免装箱
IntSummaryStatistics stats = IntStream.rangeClosed(1, 100)
    .filter(n -> n % 2 == 0)
    .summaryStatistics();

System.out.println(stats.getCount());   // 50
System.out.println(stats.getSum());     // 2550
System.out.println(stats.getAverage()); // 51.0
System.out.println(stats.getMax());     // 100

Collectors：组合才是重点03

Collectors 是 Stream 的「下游」机制：分组后还能继续计数、映射、求平均，甚至把两个收集器并联（teeing）。记住几个高频组合，比背全部 API 更有用。

record Employee(String dept, String name, int salary) {}

List<Employee> staff = List.of(
    new Employee("dev", "alice", 30),
    new Employee("dev", "bob", 20),
    new Employee("ops", "carol", 25));

Map<String, Long> countByDept = staff.stream()
    .collect(Collectors.groupingBy(Employee::dept, Collectors.counting()));

Map<String, List<String>> namesByDept = staff.stream()
    .collect(Collectors.groupingBy(Employee::dept,
        Collectors.mapping(Employee::name, Collectors.toList())));

Map<String, Double> avgByDept = staff.stream()
    .collect(Collectors.groupingBy(Employee::dept,
        Collectors.averagingInt(Employee::salary)));

System.out.println(countByDept);  // {ops=1, dev=2}
System.out.println(namesByDept);  // {ops=[carol], dev=[alice, bob]}
System.out.println(avgByDept);    // {ops=25.0, dev=25.0}

// teeing：一次遍历同时求和与计数，再算平均
record Stats(long count, int sum) {}
Stats stats = staff.stream().collect(Collectors.teeing(
    Collectors.counting(),
    Collectors.summingInt(Employee::salary),
    Stats::new));
System.out.println(stats); // Stats[count=3, sum=75]

Optional：表达「可能没有」04

Optional 是返回值的容器，不是字段、参数或集合元素的替代品。它让调用方在类型上看到「可能为空」，但不应被滥用成到处 get()。

Optional<String> name = findName(id);

// 推荐：提供默认值或做映射
String display = name.orElse("匿名");
int length = name.map(String::length).orElse(0);
name.ifPresentOrElse(
    n -> System.out.println("找到 " + n),
    () -> System.out.println("未找到"));

// 不推荐：把 Optional 当 null 用
if (name.isPresent()) {
    System.out.println(name.get());
}

• 不要用 Optional.of(null)，会直接 NPE；允许 null 用 ofNullable。
• orElse 的参数会立即求值；需要惰性计算时用 orElseGet。
• map 返回 Optional<U>，flatMap 用于函数本身返回 Optional 的场景。
• 集合元素、Map value、方法参数通常不用 Optional；返回值可以用。

Optional.of("a").orElseThrow(() -> new IllegalStateException("missing"));
Optional.ofNullable(null).orElseGet(() -> "fallback");
Optional.of("x").stream().count(); // 0 或 1，便于和 Stream 衔接

并行流：默认不要用05

parallelStream() 使用公共 ForkJoinPool，看起来一行就能并行，但它只适合数据量大、CPU 密集、无共享状态、易拆分的任务。I/O 密集、数据量小、有顺序依赖时，它通常更慢。

• 数据量小（几千个元素以内）时，拆分与合并的开销超过收益。
• forEach 不保证顺序；需要顺序用 forEachOrdered，但这会削弱并行收益。
• 有状态 lambda（修改外部集合、计数器）会导致数据竞争或错误结果。
• 公共 ForkJoinPool 的线程数有限，阻塞任务会拖垮其他并行流。
• 需要自定义线程池时，用 ForkJoinPool 提交，而不是在流里阻塞。

// 反例：有共享可变状态
List<Integer> out = new ArrayList<>();
IntStream.range(0, 1000).parallel().forEach(out::add); // 线程不安全

// 正例：无状态、可结合
long count = IntStream.range(0, 1_000_000)
    .parallel()
    .filter(n -> n % 3 == 0)
    .count();

// 正例：需要顺序时明确说明
List<Integer> ordered = IntStream.range(0, 1000)
    .parallel()
    .boxed()
    .collect(Collectors.toList());

常见坑清单06

• Stream 不能复用：终端操作后再次使用抛 IllegalStateException。
• 惰性不是不执行：忘记终端操作，整条流水线不会运行。
• sorted 是全量缓冲：对无限流使用 sorted 会永远不返回。
• flatMap 返回 null：函数返回 null 会导致 NPE；返回 Stream.empty()。
• 装箱开销：大量数值计算用 mapToInt / IntStream。
• findFirst vs findAny：并行流中 findAny 更快，但不保证是第一个。
• Collectors.toList() 不保证可变性：JDK 16 的 Stream.toList() 是明确不可变的。
• 不要在流中修改源集合：可能抛 ConcurrentModificationException。

// flatMap 正确返回空流
List<String> words = List.of("a b", "", "c d");
List<String> tokens = words.stream()
    .flatMap(s -> s.isBlank() ? Stream.empty() : Arrays.stream(s.split(" ")))
    .toList();
System.out.println(tokens); // [a, b, c, d]

// 用 Stream 生成 Map 时处理重复 key
Map<Integer, String> byLength = Stream.of("a", "bb", "cc")
    .collect(Collectors.toMap(
        String::length,
        s -> s,
        (a, b) -> a + "," + b));       // 合并重复 key
System.out.println(byLength); // {1=a, 2=bb,cc}

版本边界BOUNDARY

这张表比 API 本身更值得记。写代码时先确认你的运行时版本。

费曼自检SELF CHECK

先合上页面，用自己的话回答，再展开对照。能讲出因果和反例，才算真正掌握。