[Java/并发编程] 深度解析：Java 并行流(parallelStream) [JDK8-]

序

• 项目中利用了Java 8 的并行流(parallelStream)来优化程序处理性能：
  

1.     public static LinkedList<CycleCanSequenceDto> batchParseCloudMessageToCycleSequences(
2.         List<byte []> cloudMessageBytesList
3.         , CanHeaderConfigDto cloudMessageHeaderConfig
4.     ) {
5.         List<LinkedList<CycleCanSequenceDto>> cycleCanSequenceDtoListList = cloudMessageBytesList.parallelStream().map(cloudMessageBytes -> {//并行处理
6.             LinkedList<CycleCanSequenceDto> canSequenceDtos = null;
7.             try {
8.                 canSequenceDtos = parseCloudMessageToCycleSequences(cloudMessageBytes, cloudMessageHeaderConfig);
9.             } catch (IOException e) {
10.                 String errorMessage = "Parse cloud message to cycle sequences fail!cloudMessageBytesHex:" + BytesUtils.bytesToHexString(cloudMessageBytes);
11.                 log.error( errorMessage );
12.                 throw new RuntimeException(errorMessage);
13.             }
14.             return canSequenceDtos;
15.         } ).collect(Collectors.toCollection(LinkedList::new));
17.         LinkedList<CycleCanSequenceDto> cycleCanSequenceDtoList = cycleCanSequenceDtoListList.parallelStream().flatMap(cycleCanSequenceDtoListElement -> {//并行处理
18.             return cycleCanSequenceDtoListElement.stream();
19.         }).collect(Collectors.toCollection(LinkedList::new));
20.         return cycleCanSequenceDtoList;
21.     }

复制代码

1. @Setter
2. @Getter
3. public class CycleCanSequenceDto extends CycleMessageSequenceDto {
4.     /**
5.      * 获取 MessagePayloadDto 的总个数
6.      * @param cycleCanSequences
7.      * @return
8.      */
9.     public static Long getMessagePayloadSize(List<CycleCanSequenceDto> cycleCanSequences){
10.         AtomicLong messagePayloadSize = new AtomicLong(0);
11.         if(cycleCanSequences==null) {
12.             return -1L;
13.         }
14.         cycleCanSequences.parallelStream().forEach(cycleCanSequenceDto -> {
15.             Integer currentCycleCanSequenceDtoMessagePayloadSize = cycleCanSequenceDto.getContent().size();
16.             messagePayloadSize.addAndGet( currentCycleCanSequenceDtoMessagePayloadSize );
17.         });
18.         return messagePayloadSize.get();
19.     }
20. }

复制代码

概述：Java 并行流(parallelStream)[JDK8 - ]

• 并行流(parallelStream)是Java 8引入的强大特性，它能够自动将流操作【并行化】，以利用多核处理器的优势。
  

java.util.Collection#parallelStream()
Java 8引入了流的概念去对数据进行复杂的操作，而且使用并行流（Parallel Steams）支持并发，大大加快了运行效率。

• 与【并行流】对应的是【顺序流】
  

1. //顺序流
2. list.stream()
3.         .filter(i -> i > 10)
4.         .collect( Collectors.toList() );
7. //并行流
8. list.parallelStream()
9.         .filter(i -> i > 10)
10.         .collect( Collectors.toList() );

复制代码

下面我们将全面探讨parallelStream的使用方法、原理和最佳实践。

并行流基础

创建并行流

1. // 从集合创建并行流
2. List<String> list = Arrays.asList("a", "b", "c");
3. Stream<String> parallelStream = list.parallelStream();
5. // 将顺序流转为并行流
6. Stream<String> parallelStream2 = Stream.of("a", "b", "c").parallel();

复制代码

基本使用示例

1. List<Integer> numbers = IntStream.rangeClosed(1, 100).boxed().collect(Collectors.toList());
3. // 并行计算平方和
4. long sum = numbers.parallelStream()
5.         .mapToLong(i -> i * i)
6.         .sum();

复制代码

并行流工作原理

底层机制

• 并行流使用Fork/Join框架实现：
  

• 将任务分割为多个子任务(fork)
  
• 并行执行这些子任务
  
• 合并结果(join)
  

算法思想： 分治

• 案例讲解: 以代码list.parallelStream().filter(...).collect(...)为例
  

• Stage链构建：通过Head节点（Stage0）和中间操作（如filter、sorted）形成双向链表，每个阶段（Stage）封装操作逻辑。
  
• 任务拆分：Spliterator将数据分割为多个子任务，分发到ForkJoinPool的线程队列。
  
• 并行执行：各线程独立处理子任务，通过opWrapSink方法将操作链应用到数据流。
  
• 结果合并：终端操作（如collect）调用combiner合并子任务结果。
  

1. // 示例：ArrayList的Spliterator实现
2. public Spliterator<E> spliterator() {
3.     return new ArrayListSpliterator<>(this, 0, -1, 0); // 初始范围[0, size)
4. }

复制代码

底层框架：Fork/Join 框架

• 并行流基于Java 7引入的Fork/Join框架实现，其核心是ForkJoinPool线程池，采用工作窃取算法（Work-Stealing） 优化任务分配。
  

每个线程维护一个双端队列，优先处理自己的任务，空闲时窃取其他线程队列尾部的任务，最大化CPU利用率。

• 关键类分析：
  

• ForkJoinTask：任务基类，子类包括RecursiveTask（有返回值）和RecursiveAction（无返回值）。
  
• Spliterator：数据拆分器，负责将数据源分割为可并行处理的子块。
  

例如: ArrayListSpliterator支持高效随机访问分割。

源码级关键机制解析

1) 数据拆分与合并

• Spliterator特性：通过characteristics()方法返回特性值（如ORDERED、SIZED），影响拆分策略。
  

例如: ArrayList支持高效平均分割，而LinkedList拆分成本高。

• 任务链构造：中间操作（如filter、map）通过StatelessOp或StatefulOp节点构建操作链，StatefulOp（如sorted）需缓存中间数据。
  

2) 并行流线程模型

• 默认线程池：使用ForkJoinPool.commonPool() (JVM内共享的公共线程池, 被【整个应用程序】所使用)
  

• 默认的线程数为: Runtime.getRuntime().availableProcessors() - 1 即: CPU核心数-1。
  

-1是因为还有 JVM 的主线程需要占用1个线程

• 可自定义系统属性： java.util.concurrent.ForkJoinPool.common.parallelism
  

最佳实践: 由于主线程也会参与任务抢占CPU，所以 ForkJoinPool.commonPool 的线程数尽量设置为 （CPU核心数*N - 1）

1. // 设置全局并行度
2. System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");

复制代码

• 自定义线程池：可通过自定义ForkJoinPool提交任务，但需注意避免资源竞争。
  

支持通过 ForkJoinPool 定义私有线程池:

1. ForkJoinPool forkJoinPool = new ForkJoinPool(8);
2. List<Long> longs = forkJoinPool.submit(() -> aList.parallelStream().map( e -> {
3.     return e + 1;
4. }).collect(Collectors.toList())).get();

复制代码

适用场景

适合使用并行流的场景

• 数据量大：通常超过10,000个元素
  
• 计算密集型操作(CPU)：如复杂的数学运算
  
• 无状态操作：如map、filter、flatMap等
  
• 独立操作：元素处理不依赖其他元素
  

不适合的场景

• 顺序依赖操作：如limit、findFirst等
  
• 有状态操作：如sorted、distinct
  
• I/O密集型操作：可能导致线程阻塞 (补充意见：但也不绝对不适合，有些情况下顺序执行，反而更慢)
  
• 小数据集：并行开销可能超过收益
  

性能优化技巧

正确测量性能

1. long start = System.nanoTime();
2. result = list.parallelStream().[...].collect(Collectors.toList());
3. long duration = (System.nanoTime() - start) / 1_000_000;
4. System.out.println("耗时: " + duration + " ms");

复制代码

选择合适的并行度

1. // 自定义线程池
2. ForkJoinPool customPool = new ForkJoinPool(4);
3. customPool.submit(() -> {
4.     list.parallelStream().[...].collect(Collectors.toList());
5. }).get();

复制代码

避免共享可变状态

1. // 错误示例 - 存在竞态条件
2. List<String> result = new ArrayList<>();
3. list.parallelStream().forEach(s -> result.add(s.toUpperCase()));  // 可能抛出异常
5. // 正确做法
6. List<String> safeResult = list.parallelStream()
7.         .map(String::toUpperCase)
8.         .collect(Collectors.toList());

复制代码

高级应用

自定义Spliterator

1. class CustomSpliterator<T> implements Spliterator<T> {
2.     // 实现方法...
3. }
5. Spliterator<String> spliterator = new CustomSpliterator<>(data);
6. Stream<String> parallelStream = StreamSupport.stream(spliterator, true);

复制代码

并行收集器

1. // 使用线程安全的收集器
2. Map<String, List<Student>> studentsByClass = students.parallelStream()
3.     .collect(Collectors.groupingByConcurrent(Student::getClassName));

复制代码

FAQ: 并行流的常见陷阱与解决方案

Q:并行流与顺序流的性能对比?

• 测试示例
  

1. List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000).boxed().collect(Collectors.toList());
3. // 顺序流
4. long seqTime = measureTime(() -> numbers.stream().reduce(0, Integer::sum));
6. // 并行流
7. long parTime = measureTime(() -> numbers.parallelStream().reduce(0, Integer::sum));
9. System.out.println("顺序流: " + seqTime + "ms");
10. System.out.println("并行流: " + parTime + "ms");

复制代码

• 对比结果
  

操作数据量顺序流耗时并行流耗时求和100万15ms8ms过滤1000万120ms45ms排序100万650ms750msQ:线程安全问题

• 问题
  

1. int[] counter = new int[1];
2. list.parallelStream().forEach(e -> counter[0]++);  // 竞态条件

复制代码

• 解决
  

1. // 使用原子类
2. AtomicInteger counter = new AtomicInteger();
3. list.parallelStream().forEach(e -> counter.incrementAndGet());
5. // 或使用归约操作
6. int sum = list.parallelStream().mapToInt(e -> 1).sum();

复制代码

Q:顺序敏感操作

• 问题
  

1. // 并行流中findFirst可能不如预期
2. Optional<Integer> first = list.parallelStream()
3.         .filter(i -> i > 10)
4.         .findFirst();

复制代码

• 解决
  

1. // 如需顺序保证，使用【顺序流】，而非并行流
2. Optional<Integer> first = list.stream()
3.         .filter(i -> i > 10)
4.         .findFirst();

复制代码

Q:性能层面的考量：是否需要单独构建线程池？

• ✅ 建议单独构建线程池的场景
  | 场景            | 原因                                                                 |
  | ------------- | ------------------------------------------------------------------ |
  | I/O 密集型任务 | 默认线程数较少（CPU-1），不适合阻塞操作（如 DB、HTTP），容易拖慢整个 commonPool()，影响其他并行任务 。 |
  | 任务隔离需求    | 避免与其他模块共享线程池，防止任务间资源竞争、死锁或阻塞 。                                     |
  | 需要精确控制并发度 | 自定义线程池可设置合适的线程数，避免过度切换或资源浪费 。                                      |
  
• ❌ 可不单独构建线程池的场景
  

场景原因CPU 密集型任务默认 commonPool() 的线程数已接近 CPU 核心数，适合计算密集型任务 。简单一次性任务代码简洁、无需复杂控制，使用默认线程池即可 。Q:最佳实践经验

• 先测试后优化：不要假设并行一定更快，实际测量性能
  
• 避免副作用：确保lambda表达式没有副作用
  
• 考虑顺序性：需要顺序保证时使用顺序流
  
• 合理设置并行度：根据CPU核心数和任务特性调整
  
• 注意数据结构：ArrayList比LinkedList更适合并行处理
  
• 避免自动装箱：使用原始类型流(IntStream等)提升性能
  
• 是否需要单独创建线程池来执行并行流？
  

• CPU 密集型任务：可直接使用 parallelStream()，无需额外线程池。
  
• I/O 密集型或关键业务：建议如下方式使用自定义 ForkJoinPool：
  

若任务为 I/O 密集型或对隔离性、并发度有要求，有必要单独构建线程池以提升性能与稳定性 。

1. ForkJoinPool customPool = new ForkJoinPool(20); // 自定义线程数
2. customPool.submit(() ->
3.     list.parallelStream().forEach(item -> doSomething(item))
4. ).get();
5. customPool.shutdown();

复制代码

并行流是强大的工具，但需要谨慎使用。正确使用时可以显著提升性能，错误使用则可能导致潜在问题。理解其工作原理和适用场景是有效使用并行流的关键。

X 参考文献

• Java并行流(parallelStream)深度解析 - CSDN
  
• Java8并行流parallelStream原理深度解析 - CSDN
  
• 如何自定义ForkJoinPool提升并行流 ParallelStream执行速度 - 亿速云/大数据
  
• java8中修改parallelStream默认并发数 - CSDN
  

    本文作者：        千千寰宇   
    本文链接：         https://www.cnblogs.com/johnnyzen   
    关于博文：评论和私信会在第一时间回复，或直接私信我。   
    版权声明：本博客所有文章除特别声明外，均采用 BY-NC-SA     许可协议。转载请注明出处！
    日常交流：大数据与软件开发-QQ交流群: 774386015        【入群二维码】参见左下角。您的支持、鼓励是博主技术写作的重要动力！   

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！