8.7 Stream API 与集合流水线

前置知识：Lambda、方法调用；本页先使用 List，集合细节在第10章展开

预计时间：30 分钟

运行环境：Java 17

完成标志：能建立并解释一条无副作用的 Stream 流水线

名单不该被临时变量淹没

例如，从一批参与者中找出成年人，给姓名加上组名，再排序取前五个。这里把 List<Participant> 理解为按顺序保存参与者的列表；Participant 的数据类定义见完整示例。Stream API 把查询写成一条流水线：数据从来源进入，依次经过中间操作，最后由终端操作取出结果。

List<String> names = participants.stream()
        .filter(participant -> participant.age() >= 18)
        .map(participant -> "[" + participant.team() + "] " + participant.name())
        .sorted()
        .limit(5)
        .toList();

Stream 不是集合的新名字。集合保存数据；Stream 提供一次处理过程，而不是可供反复读取的容器。不过，排序等有状态操作可能在处理期间缓存元素。

1. 一条流水线有三部分

来源               中间操作                         终端操作
participants.stream() -> filter -> map -> sorted -> limit -> toList

部分	作用	常见例子
来源	提供元素	list.stream()、Stream.of(...)、Files.lines(...)
中间操作	返回另一条 Stream，组合处理步骤	filter、map、distinct、sorted、limit
终端操作	触发遍历并产生结果或副作用	toList、collect、count、reduce、forEach

中间操作都是惰性的。调用 filter 和 map 时只是在搭流水线，没有终端操作就不会开始遍历：

Stream<String> pipeline = List.of("铁", "木", "粮")
        .stream()
        .map(item -> {
            System.out.println("加工：" + item);
            return item + "箱";
        });

System.out.println("流水线已搭好");
long count = pipeline.count();
System.out.println("箱数：" + count);

这里 List 的大小已知，map 不改变元素数量，所以 count 可以直接取得大小，跳过整个 map；你可能只看到“流水线已搭好”和“箱数：3”。不要依赖中间操作里的打印次数来实现业务逻辑。Stream 实现可以短路，有些终端操作还可能不需要执行你以为会执行的全部步骤。

2. 方法引用是有目标类型的简写

当 Lambda 只是把参数转交给现有方法时，可以考虑方法引用：

Consumer<String> printer = System.out::println;       // 特定对象的实例方法
Function<String, Integer> parser = Integer::parseInt; // 静态方法
Function<String, Integer> length = String::length;    // 任意 String 实例的方法
Supplier<List<String>> factory = ArrayList::new;      // 构造方法

四种形式分别是 对象::实例方法、类::静态方法、类::实例方法 和 类::new。它们仍需依赖目标类型完成重载解析；方法引用并不总比 Lambda 清楚。参数需要调整、补充或校验时，直接写 Lambda 往往更直观。

3. 常用中间操作

List<Integer> numbers = List.of(3, 1, 4, 1, 5, 9, 2, 6);

List<Integer> result = numbers.stream()
        .filter(number -> number % 2 != 0) // 3, 1, 1, 5, 9
        .distinct()                        // 3, 1, 5, 9
        .sorted()                          // 1, 3, 5, 9
        .map(number -> number * 10)        // 10, 30, 50, 90
        .toList();

• filter 保留满足谓词的元素；
• map 把每个元素转换成另一个值；
• flatMap 把每个元素产生的多个值摊平成一条流；
• distinct 按 equals/hashCode 去重；
• sorted 需要自然顺序或显式 Comparator；
• limit、skip 截取序列；
• takeWhile、dropWhile 针对有序流处理连续前缀。

sorted 可能需要缓存全部元素，而 filter 通常可以逐个处理。大数据流水线里，先做能明显减少数据量的过滤，经常更省资源；但要以语义正确为前提。

4. 终端操作与空结果

long adultCount = participants.stream()
        .filter(participant -> participant.age() >= 18)
        .count();

boolean hasSenior = participants.stream()
        .anyMatch(participant -> participant.age() >= 65);

long totalAge = participants.stream()
        .mapToLong(Participant::age)
        .sum();

double averageAge = participants.stream()
        .mapToInt(Participant::age)
        .average()
        .orElse(0.0);

findFirst、没有初始值的 reduce、average 等操作可能没有结果，因此返回 Optional、OptionalInt 或 OptionalDouble。调用方要明确空输入的业务含义，不能无条件 get()。

上面的年龄总和先转为 long，避免 int 求和较早溢出；long 也有范围上限，流不会自动改用任意精度运算。处理数字时优先考虑 mapToInt、mapToLong、mapToDouble 产生的原始类型流。它们提供 sum、average 和 summaryStatistics，也避免不必要的装箱。

5. 收集结果时看清契约

Java 16 起，Stream.toList() 返回不可修改的列表，但不深复制元素，也不会让可变元素自动不可变：

List<String> names = participants.stream()
        .map(Participant::name)
        .toList();

// names.add("新用户"); // UnsupportedOperationException

若明确需要可变结果，可写：

List<String> mutableNames = participants.stream()
        .map(Participant::name)
        .collect(Collectors.toCollection(ArrayList::new));

不要依赖 Collectors.toList() 的具体实现或可变性；其契约没有保证列表类型和可变性。

收集成 Map 时还要处理重复键：

Map<String, Integer> oldestByTeam = participants.stream()
        .collect(Collectors.toMap(
                Participant::team,
                Participant::age,
                Integer::max));

没有第三个“合并函数”时，重复组名会抛 IllegalStateException。合并策略必须符合业务含义，不能为了消除异常随便保留一个值。

6. Stream 只能消费一次

Stream<String> stream = List.of("状态", "天气").stream();
long count = stream.count();
// stream.findFirst(); // IllegalStateException

终端操作完成后，这条 Stream 已被消费，不能再次使用；这不等于它持有的 I/O 资源已自动关闭。实现能检测到重复使用时会抛 IllegalStateException，API 并不保证检测每一种违规重用。需要再次遍历时，从集合重新调用 stream()；若来源本身只能读取一次，就先决定是否应物化为集合。

来自 I/O 的 Stream 还可能持有资源。例如 Files.lines(path) 应用 try-with-resources 关闭：

try (Stream<String> lines = Files.lines(path)) {
    long nonBlank = lines.filter(line -> !line.isBlank()).count();
    System.out.println(nonBlank);
}

7. 行为参数应尽量无副作用

传给 filter、map 等操作的行为应满足两条重要约束：

• 不干扰来源：遍历时不要修改正在作为来源的集合；
• 尽量无状态：不要让结果依赖会变化的外部状态。

下面的写法把业务日志塞进 peek，看似清楚，实际并不可靠：

orders.stream()
        .peek(order -> auditService.record(order)) // 不要承担必须发生的业务动作
        .map(Order::total)
        .findFirst();

短路操作可能只处理部分元素，某些优化也可能跳过不影响结果的阶段。peek 适合临时观察流水线，不适合必须执行的审计、计费或写库。必要副作用应放在边界清晰的命令式代码或专门终端处理阶段。

8. 并行流不是“免费加速”

常见 JDK 实现从普通调用线程执行并行流时，通常借助公共 ForkJoinPool；Stream API 不承诺固定的执行器，具体执行上下文也会影响所用线程池。是否变快取决于数据规模、拆分成本、每个元素的计算量、合并成本、可用核心以及同进程中的其他任务。

int total = numbers.parallelStream()
        .mapToInt(Integer::intValue)
        .sum();

归约操作需要满足结合性，行为要能安全并发执行；共享可变计数器会产生竞态。forEach 在并行流中也不保证遭遇顺序，确实需要顺序时可了解 forEachOrdered，但顺序约束可能抵消并行收益。

初学阶段默认用顺序流。真实项目若考虑并行流，应先确认语义，再用代表性负载测量，并了解公共线程池与应用其余部分的相互影响。

9. 可运行的用户报告

将下面内容保存为 StreamDemo.java：

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;

public class StreamDemo {
    public static void main(String[] args) {
        List<Participant> participants = List.of(
                new Participant("张三", 25, "基础组"),
                new Participant("李四", 17, "基础组"),
                new Participant("王五", 30, "算法组"),
                new Participant("赵六", 22, "基础组"));

        List<String> adultNames = participants.stream()
                .filter(participant -> participant.age() >= 18)
                .sorted(Comparator.comparing(Participant::name))
                .map(participant -> "[" + participant.team() + "] " + participant.name())
                .toList();

        Map<String, Integer> oldestByTeam = participants.stream()
                .collect(Collectors.toMap(
                        Participant::team,
                        Participant::age,
                        Integer::max,
                        TreeMap::new));

        double averageAge = participants.stream()
                .mapToInt(Participant::age)
                .average()
                .orElse(0.0);

        adultNames.forEach(System.out::println);
        System.out.println("各组最高年龄：" + oldestByTeam);
        System.out.printf(java.util.Locale.ROOT, "平均年龄：%.1f%n", averageAge);
    }

    record Participant(String name, int age, String team) { }
}

运行方式：

javac --release 17 -encoding UTF-8 StreamDemo.java
java StreamDemo

预期输出：

[基础组] 张三
[算法组] 王五
[基础组] 赵六
各组最高年龄：{基础组=25, 算法组=30}
平均年龄：23.5

这里显式收集到 TreeMap，所以键按组名的 String 自然顺序排列；这不是拼音或语言习惯排序。若不指定有序 Map，就不应把某次运行中的键排列写进测试契约。

离开这一页前

拿一条小流水线标出来源、中间操作与终端操作，并确认没有终端操作时不会主动遍历。终端操作后再次消费同一 Stream 应失败；空平均值、重复 Map 键和 toList() 的不可修改结果都要写明策略。业务审计不能藏进 peek，并行流也必须先证明语义正确再测量。

→ 下一步：Lambda 与 Stream 实验

API 入口：Java 17 Stream 包文档。