Stream的并行流是如何实现的？

在 Java 8 中，`Stream` 提供了并行流（Parallel Stream）来提升数据处理的效率。并行流的核心是 **ForkJoin 框架**，它能够利用多核 CPU 进行数据并行计算。 ## 1. **并行流的实现原理** 并行流的实现依赖于 `ForkJoinPool`，它是 Java 7 引入的 **工作窃取算法（Work-Stealing Algorithm）** 线程池，用于拆分任务并在多个线程上并行执行。 ### **1.1 ForkJoinPool 线程池** - `Stream` 并行计算使用 `ForkJoinPool.commonPool()` 作为默认线程池，默认线程数为 **CPU 核心数 - 1**（`Runtime.getRuntime().availableProcessors() - 1`）。 - `ForkJoinPool` 会将任务拆分成子任务，分配到多个线程执行，最后再合并结果。 ### **1.2 并行流的底层实现** 并行流的实现涉及 `Spliterator`（可分割迭代器），它会递归拆分数据，并在 `ForkJoinPool` 线程池中执行。 **核心流程** 1. **拆分数据**：`Spliterator` 负责将数据拆分成多个子任务（一般采用二分法）。 2. **任务分配**：ForkJoinPool 调用 `ForkJoinTask` 处理每个子任务。 3. **并行执行**：每个线程执行子任务，处理流中的元素。 4. **结果合并**：执行完毕后，合并计算结果。 ## 2. **示例代码** ```java import java.util.Arrays; import java.util.List; public class ParallelStreamExample { public static void main(String[] args) { List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); int sum = numbers.parallelStream() .mapToInt(Integer::intValue) .sum(); System.out.println("Sum: " + sum); } } ``` 这里的 `.parallelStream()` 会自动使用 `ForkJoinPool` 进行并行计算。 ## 3. **并行流 vs. 串行流** | 特性 | 串行流 (`stream()`) | 并行流 (`parallelStream()`) | |--------------|------------------|------------------| | 任务执行方式 | 单线程顺序执行 | 多线程并行执行 | | 适用场景 | 数据量小，操作简单 | 数据量大，计算密集型 | | 线程池 | 无 | `ForkJoinPool.commonPool()` | | 结果顺序 | 保持顺序 | 可能无序 | ## 4. **自定义 ForkJoinPool 线程数** 默认情况下，`parallelStream()` 使用 **公共的 ForkJoinPool**，但可以自定义线程池： ```java import java.util.List; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.RecursiveTask; import java.util.stream.IntStream; public class CustomForkJoinPoolExample { public static void main(String[] args) { ForkJoinPool customPool = new ForkJoinPool(4); // 指定 4 个线程 List<Integer> numbers = IntStream.range(1, 100).boxed().toList(); int result = customPool.submit(() -> numbers.parallelStream().mapToInt(Integer::intValue).sum()).join(); System.out.println("Sum: " + result); } } ``` 这样可以在特定情况下手动控制线程池的大小，避免 `commonPool` 影响其他并行任务。 ## 5. **注意事项** ### **5.1 数据拆分** - 数据量 **小** 时，`parallelStream()` 可能比 `stream()` **更慢**，因为 **线程切换的开销** 比实际计算大。 - 数据结构的 `Spliterator` 需要 **支持高效拆分**，如 `ArrayList` 适合并行，而 `LinkedList` 拆分性能较差。 ### **5.2 线程安全** - 并行流 **不保证操作的线程安全**，如果在流中修改共享变量，可能导致数据不一致： ```java List<Integer> list = new ArrayList<>(); IntStream.range(1, 1000).parallel().forEach(list::add); // 非线程安全 ``` - 可以使用 **线程安全的数据结构**（如 `ConcurrentLinkedQueue`）或 **collect** 方法： ```java List<Integer> list = IntStream.range(1, 1000) .parallel() .boxed() .collect(Collectors.toList()); // 线程安全 ``` ### **5.3 任务拆分不均** 如果数据拆分不均匀，可能导致 **负载不均衡**，某些线程长时间处于忙碌状态，而其他线程已完成任务。 ## 6. **总结** - 并行流通过 **ForkJoinPool** 进行多线程并行计算。 - `Spliterator` 负责数据拆分，使任务能够并行执行。 - 适用于 **计算密集型任务**，但可能导致额外的开销，不适用于所有场景。 - 可使用 `ForkJoinPool` 自定义线程池大小，避免默认 `commonPool` 影响其他任务。 - 并行流可能会 **影响线程安全**，需谨慎处理共享数据。 **适用场景** ✔️ 数据量大 ✔️ 计算密集型任务 ✔️ 数据结构支持高效拆分 (`ArrayList`, `IntStream.range`) **避免使用** ❌ 数据量小，额外的线程开销可能降低性能 ❌ 任务依赖顺序执行 ❌ 存在线程安全问题 并行流能提高性能，但使用时需要根据实际情况权衡是否合适。