Java中parallelStream并行流使用指南
Java中parallelStream并行流使用指南
在 Java 中，parallelStream() 是 Java 8 引入的一个用于并行处理集合数据的工具，它基于 Fork/Join框架 实现，能够自动将任务拆分成子任务并利用多核处理器并行执行。以下是对 parallelStream的详细说明和注意事项：
1. 基本用法
parallelStream() 是 Collection接口的默认方法，可以直接调用：
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
// 使用 parallelStream 并行处理
long count = numbers.parallelStream()
.filter(n -> n % 2 == 0)
.count();
2. 并行流 vs 顺序流
- 顺序流 (
stream())：单线程按顺序处理数据。 - 并行流 (
parallelStream())：数据被分成多个块，并行处理后合并结果。
3. 注意事项
(1) 线程安全问题
- 并行流的操作（如 forEach、reduce）可能涉及多线程，需确保共享变量是线程安全的。
// 错误示例：非线程安全的累加
List<Integer> unsafeList = new ArrayList<>();
numbers.parallelStream().forEach(unsafeList::add); // 可能导致数据丢失或异常
// 正确做法：使用线程安全的收集器
List<Integer> safeList = numbers.parallelStream()
.collect(Collectors.toList());
(2) 避免有状态操作
- 如果操作依赖顺序或共享状态（如
sorted()、limit()），并行可能降低性能或导致错误。
// sorted() 在并行流中可能效率更低
numbers.parallelStream().sorted().forEach(...);
(3) 性能考量
- 数据量小：并行化可能因线程调度开销而更慢。
- 任务复杂度高：适合计算密集型任务（如复杂数学运算）。
- 避免阻塞操作：如 I/O 操作会浪费线程资源。
(4) 默认线程池
- 并行流使用
ForkJoinPool.commonPool()（默认线程数 = CPU 核心数）。 - 可通过系统属性调整默认线程池：
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");
4. 适用场景
- 大规模数据集处理（如过滤、映射、归约）。
- 独立计算任务（无共享状态或顺序依赖）。
- 计算密集型操作（如数学计算、加密等）。
5. 示例
(1) 并行求和
int sum = numbers.parallelStream()
.mapToInt(Integer::intValue)
.sum();
(2) 并行过滤
List<Integer> evenNumbers = numbers.parallelStream()
.filter(n -> n % 2 == 0)
.collect(Collectors.toList());
(3) 自定义线程池（Java 8+）
ForkJoinPool customPool = new ForkJoinPool(4);
customPool.submit(() ->
numbers.parallelStream()
.forEach(n -> process(n))
).get();
6. 总结
- 优势：简化并行编程，提升多核 CPU 利用率。
- 劣势：滥用可能导致线程安全问题、性能下降。
- 原则：优先使用顺序流，仅在必要时（数据量大、任务独立）使用并行流。