快捷導(dǎo)航

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

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

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

Java并行流(Parallel Stream)

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

以下是其核心概念、使用方法及注意事項(xiàng)的詳細(xì)指南：

1. 核心概念與原理

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

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

直接生成：通過(guò)集合的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. 適用場(chǎng)景與性能優(yōu)化

推薦場(chǎng)景：

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

性能陷阱：

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

4. 注意事項(xiàng)與最佳實(shí)踐

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

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

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

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

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

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

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

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

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

順序敏感操作

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

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

配置線(xiàn)程池

默認(rèn)線(xiàn)程數(shù)：

Runtime.getRuntime().availableProcessors()

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

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

5. 性能對(duì)比示例

// 順序流 vs 并行流（處理1000萬(wàn)數(shù)據(jù)）
List<Long> numbers = LongStream.rangeClosed(1, 10_000_000)
                               .boxed().collect(Collectors.toList());

// 順序流耗時(shí)
long start = System.currentTimeMillis();
long seqSum = numbers.stream().mapToLong(n -> n * 2).sum();
System.out.println("順序流耗時(shí): " + (System.currentTimeMillis() - start) + "ms");

// 并行流耗時(shí)
start = System.currentTimeMillis();
long parSum = numbers.parallelStream().mapToLong(n -> n * 2).sum();
System.out.println("并行流耗時(shí): " + (System.currentTimeMillis() - start) + "ms");

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