大聪明教你学Java | 深入浅出聊 Stream.parallel()

前言

🍊作者简介： 不肯过江东丶，一个来自二线城市的程序员，致力于用“猥琐”办法解决繁琐问题，让复杂的问题变得通俗易懂。
🍊支持作者： 点赞👍、关注💖、留言💌~

相信各位小伙伴对 Stream 都不陌生，它是 Java 8 及以后版本中引入的一个新特性，用于处理集合数据。Stream 是对集合（Collection）对象功能的增强，与 Lambda 表达式结合，可以提高编程效率、间接性和程序可读性。Stream API 中为我们提供了很多高效且易用的方法，大聪明的好朋友 —— 大明白就对这些方法情有独钟，但是就在前几天，却因为他在项目中使用了 Stream.parallel() 而引发了一个小小的意外情况… …

这里卖个关子😝~ 在说大明白引发的意外情况之前，我们先来一起看看什么是Stream.parallel()

Stream.parallel()

Stream.parallel() 方法用于将流操作转换为并行操作，以便在多个线程上并行执行。并行流是一种可以同时在多个线程上执行操作的流，它将流的元素分割成多个子集，每个子集在不同的线程上独立处理，最后将结果合并。使用 parallel() 方法可以轻松开启并行流处理模式，无需显式管理线程和同步。

List<Integer> numbers = ...; // 假设这里有一个包含大量正整数的List集合numbers.stream() // 创建顺序流.parallel() // 转换为并行流.filter(n -> n % 2 == 0) // 并行流操作 - 过滤List集合中的偶数.map(n -> n * 2) // 并行流操作 - 将过滤出来的偶数×2.forEach(System.out::println); // 并行流操作 - 打印结果

在上面的示例中，parallel() 方法将顺序流转换为并行流，后续的 filter()、map() 和 forEach() 操作将在多个线程上并行执行，从而加速数据处理。我们下面再看看它的底层原理👇

当调用 Stream.parallel() 方法时，它实际上会返回一个新的并行流对象，这个流对象可以在多个线程上并行执行流操作。下面是 Stream.parallel() 方法的大致工作原理：
① 并行流的划分和分治：当我们对并行流进行操作时，Java 会使用 Fork/Join 框架将数据划分成多个小任务，并将这些小任务分配给多个线程来并行执行。这个过程涉及到递归地将大任务分解为小任务，直到小任务足够简单可以直接求解。
② 工作窃取（Work Stealing）：Fork/Join 框架采用工作窃取算法来实现任务的调度和执行。在工作窃取的过程中，空闲的线程会主动去其他线程的任务队列中窃取任务执行。这种方式能够充分利用线程资源，提高并行处理的效率。
③合并结果：在并行流的操作中，各个线程会并行地对数据进行处理，最后需要将各个线程的处理结果进行合并，得到最终的结果。这一过程涉及到结果的收集和合并，确保最终的结果是完整且正确的。

这里我们又引申出了一个新的概念 —— Fork/Join 框架。Fork/Join 框架是 Java 7 中引入的用于支持并行计算的框架，是一种并行计算模式，用于解决可以被分解成更小的可并行任务的问题。该模式包含两个关键操作：Fork（分解）和Join（合并）。在 Fork/Join 模式中，原始问题被递归地分解为更小的子问题，直到达到可以并行解决的最小单位。这个过程被称为 Fork。每个子问题可以独立地在不同的处理器上执行，并行地求解部分问题。一旦所有的子问题都被解决，就会进行 Join 操作。Join 操作将所有子问题的结果合并为最终的解决方案。这种分解和合并的过程可以视为树形结构，其中每个节点代表一个子问题。

Fork/Join 模式最适用于可以自然地分解为多个独立子问题的计算密集型任务。它适用于多核处理器或并行计算环境，其中可以充分利用并行性。Java 平台提供了 Fork/Join 框架，用于实现该模式。它包括了一个线程池（ForkJoinPool） 和 任务（ForkJoinTask） 的概念。任务可以是可分解的子问题，也可以是执行最终计算的任务。通过 ForkJoinPool，可以将任务提交给线程池执行，自动实现任务的分解和合并过程。Fork/Join 模式的优点在于它能够充分利用多核处理器的并行性，提高计算效率。

📌 在这里我们就先对 Fork/Join 框架做一个简单的介绍，后续大聪明会单独出一篇博客对 Fork/Join 框架进行详细的介绍。

咱们言归正传，有些小伙伴看到“线程池（ForkJoinPool）”的时候可能就已经猜测到大明白遇倒的意外情况和线程有关系了。Stream.parallel() 并行流默认使用的是 ForkJoinPool.commonPool() 作为线程池，该线程池默认最大线程数就是 CPU 核数。正是因为大明白对并行流操作的原理不清楚，他在没有配置线程池的情况下，通过并行流做了数据库的大量批量更新操作，于是最大线程数只有 CPU 核数，最终导致在批量更新的时候出现了线程阻塞的情况，从而出现了这个小小的意外。

通过这件事应该也可以给各位小伙伴提个醒，在实际使用时需要慎重考虑并行化带来的影响，并确保线程安全性和并发性。

① 线程安全：并行流并不能保证线程安全性，因此，如果流中的元素是共享资源或操作本身不是线程安全的，你需要确保正确同步或使用线程安全的数据结构。
② 资源消耗：并行流默认使用的线程池大小可能与机器的实际物理核心数相适应，但也可能与其他并发任务争夺系统资源。
③ 结果一致性：并行流并不保证执行的顺序性，也就是说，如果流操作的结果依赖于元素的处理顺序，则不应该使用并行流。
④ 事务处理：在涉及到事务操作时，通常需要避免在并行流中直接处理，如上述例子所示，应当将事务边界放在单独的服务方法内，确保每个线程内的事务独立完成。