Stream API 实战：惰性求值、短路与并行流的正确用法

Stream 把"做什么"和"怎么做"分开，写出来的代码更接近业务描述。但它的执行模型是惰性的，很多看似直观的写法会带来隐藏的遍历次数或装箱开销，并行流更不是"加上 parallel 就变快"。

本文从流水线的三个阶段讲起，配合可运行的例子说明求值时机、短路行为、收集器选择，以及并行流真正适用的条件。

流水线的三个阶段

一个 Stream 操作链由三部分组成：数据源（stream()、Arrays.stream、Stream.of）、中间操作（返回 Stream，惰性）、终止操作（返回结果，触发执行）。

List<String> result = names.stream()          // 源
        .filter(n -> n.length() > 2)          // 中间，惰性
        .map(String::toUpperCase)             // 中间，惰性
        .limit(3)                             // 中间，惰性
        .collect(Collectors.toList());        // 终止，触发执行

没有终止操作时，中间操作一行都不会执行。可以用打印验证：

Stream.of("a", "bb", "ccc")
        .filter(s -> {
            System.out.println("filter " + s);
            return true;
        });
// 没有任何输出

惰性求值与短路

中间操作分为有状态与无状态。filter、map、flatMap 是无状态的，元素可以逐个处理；sorted、distinct、limit、skip 是有状态的，需要缓冲或记录状态。

limit、findFirst、anyMatch、noneMatch 会短路：元素处理到满足条件就停止，不会遍历完整数据源。

List<Integer> nums = Arrays.asList(1, 2, 3, 4, 5, 6);
long count = nums.stream()
        .peek(n -> System.out.println("peek " + n))
        .filter(n -> n % 2 == 0)
        .count();
// 输出 peek 1 到 peek 6

注意：count() 在 JDK 9 之后如果源的大小已知且没有改变元素数量的操作，会直接返回 size，peek 不再执行。这是 JDK 8 与 9+ 的一个行为差异，依赖 peek 做副作用的代码会失效。

peek 的设计意图是调试，不是业务逻辑。JDK 文档明确说明它主要用于观察流水线，不应依赖它的执行次数。

常用操作的选择

map 一对一转换，flatMap 一对多展开：

List<List<Integer>> nested = Arrays.asList(
        Arrays.asList(1, 2), Arrays.asList(3, 4));
List<Integer> flat = nested.stream()
        .flatMap(List::stream)
        .collect(Collectors.toList());   // [1, 2, 3, 4]

reduce 做聚合，注意带初始值的版本在并行流中要求初始值是单位元：

int sum = IntStream.rangeClosed(1, 100).reduce(0, Integer::sum);
Optional<String> longest = Stream.of("a", "abc", "ab")
        .reduce((x, y) -> x.length() >= y.length() ? x : y);

collect 是最灵活的终止操作。常用收集器：

Map<Boolean, List<User>> partition = users.stream()
        .collect(Collectors.partitioningBy(u -> u.getAge() >= 18));

Map<String, Long> byCity = users.stream()
        .collect(Collectors.groupingBy(User::getCity, Collectors.counting()));

Map<String, User> index = users.stream()
        .collect(Collectors.toMap(User::getId, u -> u, (a, b) -> a));

toMap 遇到重复 key 会抛 IllegalStateException，必须提供合并函数或改用 groupingBy。groupingBy 默认返回 HashMap，需要顺序时传入 LinkedHashMap::new。

基本类型流与装箱

Stream<Integer> 的每个元素都要装箱，小整数还有缓存，大范围时会产生大量临时对象。mapToInt、mapToLong、mapToDouble 返回原始类型流，sum、average、summaryStatistics 都直接可用。

IntSummaryStatistics stats = users.stream()
        .mapToInt(User::getAge)
        .summaryStatistics();
System.out.println(stats.getMax() + " " + stats.getAverage());

反方向的 boxed() 只在需要对象时使用。JDK 的 sum() 在 IntStream 上直接累加，不会装箱。

并行流的真实条件

并行流基于 ForkJoinPool.commonPool()，默认线程数为 CPU 核数减一。它把数据源切分成多段，各段独立执行，最后合并结果。

适用条件相当苛刻：

1. 数据量大。经验值是元素处理耗时乘以数量要足够大，通常几万元素以上才可能看到收益。
2. 每个元素的计算足够重。纯内存的简单映射，切分和合并的开销会盖过收益。
3. 数据源可高效切分。ArrayList、数组、IntStream.range 支持随机访问，切分成本低；LinkedList、Stream.iterate 切分代价高。
4. 操作无状态且不依赖顺序。findFirst 在并行流中会退化为串行语义，forEach 不保证顺序，需要顺序时用 forEachOrdered。
5. 没有共享可变状态。往 ArrayList 里 add 会有线程安全问题，必须用 collect 或并发容器。

// 反例：共享可变状态
List<Integer> out = new ArrayList<>();
IntStream.range(0, 1000).parallel().forEach(out::add); // 结果不确定

// 正例
List<Integer> ok = IntStream.range(0, 1000).parallel()
        .boxed()
        .collect(Collectors.toList());

在 Web 应用中还有一个隐患：commonPool 是全局共享的，一个慢任务会拖累其他并行流。阻塞型任务（IO）绝对不要放进去，需要时自定义 ForkJoinPool 并提交任务。

自定义收集器

Collectors 提供的收集器覆盖了大多数场景，但遇到"分页收集""按条件分组后取最大"这类需求时，自定义 Collector 更清晰。Collector 由五个函数组成：Supplier（创建容器）、Accumulator（累加单个元素）、Combiner（合并两个容器）、Finisher（最终转换）、Characteristics（特性标记）。

public static <T> Collector<T, ?, List<T>> toPagedList(int pageSize, int page) {
    return Collector.of(
            ArrayList::new,
            (list, item) -> {
                if (list.size() < (page + 1) * pageSize) {
                    list.add(item);
                }
            },
            (a, b) -> {
                for (T t : b) {
                    if (a.size() < (page + 1) * pageSize) {
                        a.add(t);
                    }
                }
                return a;
            },
            list -> list.subList(
                    Math.min(page * pageSize, list.size()),
                    Math.min((page + 1) * pageSize, list.size())),
            Collector.Characteristics.CONCURRENT);
}

Characteristics.CONCURRENT 表示容器可被多线程并发累加，只在并行流下有意义，且要求 Supplier 返回的容器本身线程安全。UNORDERED 表示不关心顺序，IDENTITY_FINISH 表示 Finisher 是恒等函数，可以省略。

自定义收集器的价值在于把"多次遍历"合并成一次。上面的分页如果先用 skip 再 limit，在有序流上仍然需要遍历前面的元素；而用收集器可以在遍历一次的过程中直接定位区间。

Optional 与 Stream 的衔接

findFirst、max、min、reduce 单参版本都返回 Optional。Optional 的用途是表达"可能没有结果"这一返回值语义，不应作为字段类型或方法参数，也不应到处 isPresent() 判断，那等于把 null 检查换了个写法。

常用的组合是 map 与 orElseGet：

String city = users.stream()
        .filter(u -> u.getId().equals(id))
        .findFirst()
        .map(User::getCity)
        .orElse("unknown");

orElse 无论是否为空都会计算默认值，orElseGet 只在为空时调用 Supplier。默认值来自数据库查询或远程调用时，必须用 orElseGet，否则每次都会多一次无用请求。

Optional.stream()（JDK 9+）可以把 Optional 展开进流，适合 flatMap 过滤空值：

List<String> names = ids.stream()
        .map(this::findUser)          // Stream<Optional<User>>
        .flatMap(Optional::stream)    // 自动丢弃 empty
        .map(User::getName)
        .collect(Collectors.toList());

JDK 8 没有这个方法，等价写法是 .filter(Optional::isPresent).map(Optional::get)。

常见坑 / 最佳实践

1. 在 Stream 中修改外部集合。会产生 ConcurrentModificationException 或数据错乱，应通过 collect 返回新集合。
2. 复用 Stream。Stream 只能消费一次，第二次调用终止操作抛 IllegalStateException。
3. findFirst 返回 Optional 直接 get()。空流时抛 NoSuchElementException，用 orElse、orElseGet 或 ifPresent。
4. 用 parallelStream() 处理 IO。commonPool 被阻塞会影响全局，且线程数固定，无法应对高延迟。
5. 忽视 sorted 的稳定性与比较器空值。Comparator.nullsLast 显式处理 null，避免 NPE。
6. 过度链式导致可读性下降。超过 6 到 8 个操作时拆成有名字的中间变量或方法，调试时也更方便。
7. 在 forEach 里做副作用。日志、远程调用、数据库写入都不适合放在流里，forEach 应只用于最终消费。

一个完整的实战例子

按城市统计成年用户数量，并按数量倒序取前 3：

Map<String, Long> top3 = users.stream()
        .filter(u -> u.getAge() >= 18)
        .collect(Collectors.groupingBy(User::getCity, Collectors.counting()))
        .entrySet().stream()
        .sorted(Map.Entry.<String, Long>comparingByValue().reversed())
        .limit(3)
        .collect(Collectors.toMap(
                Map.Entry::getKey,
                Map.Entry::getValue,
                (a, b) -> a,
                LinkedHashMap::new));

这里用了两次流：第一次聚合，第二次排序取前 N。因为 groupingBy 的结果需要全量才能排序，无法在一次遍历中完成。指定 LinkedHashMap 是为了保留顺序，否则 toMap 默认返回 HashMap，顺序会丢失。

小结

• Stream 是惰性的，只有终止操作才触发执行，中间操作分为有状态与无状态。
• limit、findFirst、anyMatch 会短路，peek 只用于调试，且 JDK 9 起 count 可能跳过它。
• 基本类型流避免装箱，mapToInt 加 summaryStatistics 是统计场景的常用组合，需要对象时再 boxed()。
• 并行流只在数据量大、计算重、源可切分、无共享状态时才值得用，阻塞任务必须用自定义线程池。
• toMap 必须考虑重复 key，需要顺序时显式指定 LinkedHashMap；orElseGet 优于 orElse。