快捷導(dǎo)航

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

更新時間：2025年03月05日 16:07:16 作者：wujiada001

Java并行流（ParallelStream）通過多線程并行處理集合數(shù)據(jù),利用Fork/Join框架加速計算,適用于大規(guī)模數(shù)據(jù)集和計算密集型任務(wù),并行流主要通過集合的parallelStream()方法或現(xiàn)有流的parallel()方法創(chuàng)建,適用于數(shù)據(jù)量大、計算復(fù)雜且無狀態(tài)操作的場景

Java并行流(Parallel Stream)

并行流是Java 8引入的高效處理集合數(shù)據(jù)的工具，通過多線程加速計算。

以下是其核心概念、使用方法及注意事項的詳細指南：

1. 核心概念與原理

并行處理機制：將數(shù)據(jù)分割為多個塊，利用Fork/Join框架在多個線程上并行處理，最后合并結(jié)果。
默認線程池：使用ForkJoinPool.commonPool()，線程數(shù)等于CPU核心數(shù)（可通過系統(tǒng)參數(shù)調(diào)整）。
適用場景：大規(guī)模數(shù)據(jù)集、計算密集型任務(wù)（如數(shù)學(xué)運算、批量轉(zhuǎn)換）。

2. 創(chuàng)建并行流的方式

直接生成：通過集合的parallelStream()方法。
轉(zhuǎn)換順序流：在現(xiàn)有流上調(diào)用parallel()。

List<Integer> list = Arrays.asList(1, 2, 3, 4);

// 方式1：直接生成并行流
Stream<Integer> parallelStream1 = list.parallelStream();

// 方式2：將順序流轉(zhuǎn)為并行
Stream<Integer> parallelStream2 = list.stream().parallel();

3. 適用場景與性能優(yōu)化

推薦場景：

數(shù)據(jù)量大：如百萬級元素的過濾、映射。
計算復(fù)雜：如矩陣運算、圖像處理。
無狀態(tài)操作：如map、filter、reduce（不依賴處理順序或外部變量）。

性能陷阱：

小數(shù)據(jù)集：并行化開銷（線程調(diào)度、數(shù)據(jù)分割）可能抵消收益。
低耗時操作：如簡單加減法，并行可能更慢。

4. 注意事項與最佳實踐

避免共享可變狀態(tài)

并行操作中修改共享變量會導(dǎo)致線程安全問題，應(yīng)使用無狀態(tài)操作或同步控制。

// 錯誤示例：線程不安全的累加
List<Integer> nums = Arrays.asList(1, 2, 3);
int[] sum = {0};
nums.parallelStream().forEach(n -> sum += n); // 結(jié)果可能錯誤

// 正確做法：使用歸約
int safeSum = nums.parallelStream().reduce(0, Integer::sum);

謹慎使用有狀態(tài)操作

如sorted()、distinct()在并行流中可能更耗時，需合并線程結(jié)果。

// 并行排序（可能比順序流慢）
List<Integer> sortedList = nums.parallelStream().sorted().toList();

數(shù)據(jù)源的可拆分性

高效結(jié)構(gòu)：ArrayList、數(shù)組（支持快速隨機訪問，易于分割）。
低效結(jié)構(gòu)：LinkedList、TreeSet（拆分成本高）。

順序敏感操作

使用forEachOrdered保證順序，但犧牲性能。

// 按順序輸出（性能低于無序操作）
list.parallelStream().forEachOrdered(System.out::println);

配置線程池

默認線程數(shù)：

Runtime.getRuntime().availableProcessors()

修改全局線程數(shù)：

# JVM啟動參數(shù)
-Djava.util.concurrent.ForkJoinPool.common.parallelism=8

5. 性能對比示例

// 順序流 vs 并行流（處理1000萬數(shù)據(jù)）
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");

典型結(jié)果（8核CPU）：