Java 8 函数式编程入门：从 Lambda 到 Stream 并行执行

目录

1. Lambda 表达式：将“行为逻辑”作为参数传递

2. Stream API：基于流水线模型的数据处理机制（详解）

2.1 Stream 的核心思想：描述“数据如何被处理”

2.2 Stream 的三大组成部分

2.2.1 数据源（Source）

2.2.2 中间操作（Intermediate Operation）

2.2.3 终止操作（Terminal Operation）

2.3 惰性执行与流水线优化机制

2.4 串行流与并行流

2.4.1 串行流（Sequential Stream）

2.4.2 并行流（Parallel Stream）

a.什么是 Fork/Join 框架？

b.并行流的执行流程

c.Fork/Join 与并行流的关系

3. Optional：显式表达“值可能不存在”的语义

Java 8 引入了 Lambda 表达式、Stream API 和 Optional，为集合数据处理和空值管理提供了更加简洁且安全的编程方式。Lambda 用于描述可传递的行为逻辑，Stream API 通过流水线模型对数据进行声明式处理，而 Optional 则显式建模“可能为空”的值，从而降低空指针异常的风险。

在执行层面，Stream API 同时支持串行流与并行流。串行流强调顺序性与可预测性，并行流则基于多线程机制提升计算效率，适用于 CPU 密集型场景。二者在保持统一编程模型的同时，为不同业务需求提供了灵活的执行选择。

1. Lambda 表达式：将“行为逻辑”作为参数传递

在传统 Java 编程中，方法参数通常只能传递数据，而无法直接传递“要执行的操作”。当需要将一段处理逻辑作为参数传入方法时，往往只能通过匿名内部类的方式实现，这不仅代码冗长，也削弱了逻辑本身的可读性。Lambda 表达式的引入，正是为了解决这一问题。

这里所说的“行为逻辑”，本质上是指对某个输入执行的操作规则，例如：如何处理一个元素、如何进行判断、如何完成一次计算等。Lambda 允许将这些操作规则以函数的形式进行传递，使代码能够直接表达“对数据做什么”。

// 匿名内部类：传递行为逻辑
list.forEach(new Consumer<Integer>() {
@Override
public void accept(Integer x) {
System.out.println(x);
}
});

// Lambda：更直观地描述行为
list.forEach(x -> System.out.println(x));

Lambda 表达式必须依附于函数式接口（Functional Interface） 才能使用。所谓函数式接口，是指只包含一个抽象方法的接口，例如 Runnable、Comparator、Consumer 等。Lambda 并不是脱离接口存在的函数，而是函数式接口的简洁实现形式。

@FunctionalInterface
public interface Calculator {
int add(int a, int b);
}

// Lambda 实现函数式接口
Calculator calc = (a, b) -> a + b;

通过 Lambda，Java 在保持类型安全的前提下，实现了对“行为抽象”的简化表达，为后续 Stream API 的设计奠定了基础。

提醒：类型安全指的是

程序在编译期和运行期，能够保证变量、表达式和方法调用只会以“符合其声明类型”的方式被使用，从而避免不合法的类型操作。

一个最简单的例子：

String s = "hello";
Integer i = s; // 编译期直接报错

2. Stream API：基于流水线模型的数据处理机制（详解）

2.1 Stream 的核心思想：描述“数据如何被处理”

Stream API 并不是一种新的集合结构，而是对集合数据处理过程的一种抽象描述。它的设计目标并非替代 List 或 Set，而是将原本分散在多层循环和条件判断中的逻辑，组织为一条清晰、可组合的数据处理流水线。

// 传统写法：多层循环 + 条件判断（逻辑分散）
List<Long> result = new ArrayList<>();

for (Order order : orders) {
if (order.isPaid()) {
if (order.getAmount() > 100) {
if (!blackList.contains(order.getUserId())) {
result.add(order.getId());
}
}
}
}

从使用角度来看，Stream 更关注“对数据做什么”，而不是“如何遍历数据”。这种声明式风格使得复杂的数据处理逻辑能够以接近自然语言的方式表达。

users.stream()
.filter(u -> u.getAge() >= 18)
.map(User::getName)
.sorted()
.forEach(System.out::println);

上述代码中并未显式出现循环、索引或中间变量，但完整地描述了数据的处理规则。

2.2 Stream 的三大组成部分

从结构上看，一个完整的 Stream 由 数据源（Source）、中间操作（Intermediate Operation） 和 终止操作（Terminal Operation） 三部分构成。

2.2.1 数据源（Source）

数据源用于生成 Stream，常见来源包括集合、数组以及其他 I/O 数据结构。

list.stream();
Arrays.stream(arr);

数据源本身并不会触发任何计算，仅用于定义数据的起点。

2.2.2 中间操作（Intermediate Operation）

中间操作用于描述数据的转换规则，例如过滤、映射、排序等。中间操作的一个关键特征是惰性执行：它们只会记录操作逻辑，而不会立即执行。

Stream<Integer> s = list.stream()
.filter(x -> x > 10)
.map(x -> x * 2);

在没有终止操作之前，上述代码不会对数据进行任何实际处理。常见的中间操作包括：

• filter：筛选元素
• map / flatMap：数据转换
• distinct：去重
• sorted：排序
• limit / skip：截取数据

2.2.3 终止操作（Terminal Operation）

终止操作用于触发 Stream 的执行，并生成最终结果。一旦终止操作被调用，Stream 中定义的所有中间操作将被一次性执行。

long count = list.stream()
.filter(x -> x > 10)
.count();

常见终止操作包括 forEach、collect、count、reduce 等。需要注意的是，Stream 只能被消费一次，终止操作执行完成后，Stream 即进入关闭状态。

2.3 惰性执行与流水线优化机制

Stream API 采用惰性执行机制，其核心优势在于能够对整个操作链进行统一调度与优化。所谓惰性执行，是指 Stream 的中间操作并不会在声明时立即执行，而是延迟到终止操作触发时统一执行。这一设计使得 JVM 可以在运行时对完整的操作流水线进行分析，从而减少不必要的中间结果生成。

list.stream()
.filter(x -> x > 10)
.map(x -> x * 2)
.forEach(System.out::println);

在上述示例中，filter 与 map 均为中间操作，它们仅用于描述数据处理规则，并不会独立执行。只有在调用终止操作 forEach 后，整个 Stream 才会被真正消费。

在底层执行过程中，Stream 不会先生成一个“过滤后的集合”，再生成一个“映射后的集合”，而是采用流水线融合（Pipeline Fusion）的方式，对数据进行逐元素处理。具体而言，每个元素在一次遍历过程中依次完成 filter → map → forEach 的操作链，从而避免多次遍历和中间集合创建。

这种流水线执行模型具有以下优势：

• 减少中间对象创建，降低内存开销
• 避免多次遍历集合，提高执行效率
• 为后续的短路优化（如 findFirst、anyMatch）提供基础

通过惰性执行与流水线融合机制，Stream API 在保证代码表达能力的同时，实现了较高的运行效率。

2.4 串行流与并行流

2.4.1 串行流（Sequential Stream）

默认情况下，Stream 以串行方式执行，由单线程顺序处理数据。

list.stream()
.map(this::compute)
.forEach(System.out::println);

串行流的特点是：

• 执行顺序明确

• 行为可预测

• 线程安全风险低

因此，大多数业务场景下，串行流是首选方案。

需要注意的是，串行流虽然以单线程方式执行，但其底层仍然遵循 Stream API 统一的流水线（Pipeline）执行模型。 在串行流中，终止操作作为消费端，按需从数据源中逐个拉取元素，每个元素依次经过流水线中的各个中间操作节点完成处理。

也就是说，串行流并不是对每一个中间操作单独进行一次完整遍历，而是采用 逐元素、按操作链顺序执行 的方式完成计算。这一执行模型与并行流在逻辑结构上保持一致，差异仅体现在执行驱动方式上。

在串行流场景下：

• 由单一线程驱动整条流水线执行

• 数据元素按照数据源的遍历顺序被依次处理

• 每一时刻仅有一个元素在流水线中流转

这种设计保证了串行流在具备良好可读性的同时，仍然能够避免不必要的中间集合创建，并维持较高的执行效率。

2.4.2 并行流（Parallel Stream）

并行流可以通过 parallelStream() 或 stream().parallel() 启用，其底层并不是简单地“开几个线程同时跑”，而是基于 Fork/Join 框架 实现的一种自动并行计算模型。

list.parallelStream()
.map(this::heavyCompute)
.forEach(System.out::println);

当使用并行流时，Stream 会尝试将数据拆分为多个子任务，并在多个线程中并发执行，以提升 CPU 密集型计算场景下的整体吞吐能力。

a.什么是 Fork/Join 框架？

Fork/Join 框架是 Java 7 引入的一种并行计算框架，主要用于解决可以递归拆分的大规模任务。它的核心思想可以概括为两点：

• Fork（拆分）：将一个大任务拆分为多个规模更小的子任务
• Join（合并）：在子任务执行完成后，将结果合并为最终结果

与传统线程池不同，Fork/Join 框架采用了工作窃取（Work-Stealing）算法。每个工作线程都维护一个双端队列，当某个线程完成自身任务后，会主动“窃取”其他线程尚未完成的任务，从而提升 CPU 资源利用率。

b.并行流的执行流程

在底层执行过程中，并行流通常会经历以下几个阶段：

1️⃣ 使用 Spliterator 拆分数据 数据源对应的 Spliterator 会根据数据结构特性，将原始数据不断拆分为更小的区间，以便后续并行处理。

Spliterator 是对传统 Iterator 的增强，用于支持 Stream 尤其是并行流的数据拆分与遍历。与只能顺序遍历的迭代器不同，Spliterator 提供了 按需拆分数据源的能力，这是并行流实现的基础。

其核心能力包括：

• 顺序推进元素（类似 Iterator）
• 通过拆分操作将数据源划分为多个子区间，以支持并行处理

在并行流场景下，Spliterator 会根据数据结构特性递归拆分数据，从而为后续多线程执行提供任务划分依据。对于支持高效随机访问的数据结构（如 ArrayList），拆分成本较低，并行执行效果通常更好。

2️⃣ 将子任务提交到 ForkJoinPool.commonPool 拆分后的任务会被提交到 JVM 提供的公共 Fork/Join 线程池中，由多个工作线程并发执行。

3️⃣ 多个线程并行处理子任务 每个线程独立处理一部分数据，执行相同的 Stream 操作链（如 map、filter 等）。

4️⃣ 合并计算结果（如有需要） 对于 reduce、collect 等终止操作，框架会在子任务完成后自动合并各个线程的计算结果。

c.Fork/Join 与并行流的关系

并行流正是建立在 Fork/Join 框架之上的一种高层抽象：

• Stream 负责 定义计算规则（做什么）
• Fork/Join 负责 调度线程和执行任务（怎么并行）

并行流默认使用的是：

ForkJoinPool.commonPool

该线程池具有以下特点：

• 线程数通常为 CPU 核心数 – 1
• JVM 进程内全局共享
• 会与其他并行任务（如 CompletableFuture）竞争线程资源

3. Optional：显式表达“值可能不存在”的语义

在 Java 8 之前，null 是对象缺失的主要表达方式，但隐式的空值语义极易引发空指针异常，尤其在多层对象访问或链式调用场景中。Optional 的引入，为“值可能不存在”提供了一种显式且受约束的表达方式。

Optional 本质上是一个最多只包含一个值的容器，用于提醒调用者在使用值之前必须考虑其存在性。

// 传统判空写法
if (user != null && user.getName() != null) {
System.out.println(user.getName());
}

// Optional 写法
Optional.ofNullable(user)
.map(User::getName)
.ifPresent(System.out::println);

通过 Optional，空值处理逻辑可以与业务逻辑自然组合，使判空行为更加集中和统一，从而提升代码的健壮性和可维护性。