Java之并行流(Parallel Stream)使用详解_Java

java并行流(parallel stream)

并行流是java 8引入的高效处理集合数据的工具，通过多线程加速计算。

以下是其核心概念、使用方法及注意事项的详细指南：

1. 核心概念与原理

并行处理机制：将数据分割为多个块，利用fork/join框架在多个线程上并行处理，最后合并结果。
默认线程池：使用forkjoinpool.commonpool()，线程数等于cpu核心数（可通过系统参数调整）。
适用场景：大规模数据集、计算密集型任务（如数学运算、批量转换）。

2. 创建并行流的方式

直接生成：通过集合的parallelstream()方法。
转换顺序流：在现有流上调用parallel()。

list<integer> list = arrays.aslist(1, 2, 3, 4);

// 方式1：直接生成并行流
stream<integer> parallelstream1 = list.parallelstream();

// 方式2：将顺序流转为并行
stream<integer> parallelstream2 = list.stream().parallel();

3. 适用场景与性能优化

推荐场景：

数据量大：如百万级元素的过滤、映射。
计算复杂：如矩阵运算、图像处理。
无状态操作：如map、filter、reduce（不依赖处理顺序或外部变量）。

性能陷阱：

小数据集：并行化开销（线程调度、数据分割）可能抵消收益。
低耗时操作：如简单加减法，并行可能更慢。

4. 注意事项与最佳实践

避免共享可变状态

并行操作中修改共享变量会导致线程安全问题，应使用无状态操作或同步控制。

// 错误示例：线程不安全的累加
list<integer> nums = arrays.aslist(1, 2, 3);
int[] sum = {0};
nums.parallelstream().foreach(n -> sum += n); // 结果可能错误

// 正确做法：使用归约
int safesum = nums.parallelstream().reduce(0, integer::sum);

谨慎使用有状态操作

如sorted()、distinct()在并行流中可能更耗时，需合并线程结果。

// 并行排序（可能比顺序流慢）
list<integer> sortedlist = nums.parallelstream().sorted().tolist();

数据源的可拆分性

高效结构：arraylist、数组（支持快速随机访问，易于分割）。
低效结构：linkedlist、treeset（拆分成本高）。

顺序敏感操作

使用foreachordered保证顺序，但牺牲性能。

// 按顺序输出（性能低于无序操作）
list.parallelstream().foreachordered(system.out::println);

配置线程池

默认线程数：

runtime.getruntime().availableprocessors()

修改全局线程数：

# jvm启动参数
-djava.util.concurrent.forkjoinpool.common.parallelism=8

5. 性能对比示例

// 顺序流 vs 并行流（处理1000万数据）
list<long> numbers = longstream.rangeclosed(1, 10_000_000)
                               .boxed().collect(collectors.tolist());

// 顺序流耗时
long start = system.currenttimemillis();
long seqsum = numbers.stream().maptolong(n -> n * 2).sum();
system.out.println("顺序流耗时: " + (system.currenttimemillis() - start) + "ms");

// 并行流耗时
start = system.currenttimemillis();
long parsum = numbers.parallelstream().maptolong(n -> n * 2).sum();
system.out.println("并行流耗时: " + (system.currenttimemillis() - start) + "ms");

典型结果（8核cpu）：