java并行流處理具體方案講解
更新時間：2026年02月04日 10:33:41 作者：~?？
并行流是Java?8引入的一個強大工具,可以幫助我們充分利用多核處理器的性能,加速數(shù)據(jù)處理過程,這篇文章主要介紹了java并行流處理的相關資料,文中通過代碼介紹的非常詳細,需要的朋友可以參考下
當我們需要調(diào)用多次調(diào)用第三方接口，我們采用并行流處理的方式發(fā)送請求。
實際場景：第三方給了一個表，我需要對遍歷整個表，調(diào)用表中的所有地址并解析json獲取數(shù)據(jù)
常規(guī)思維：如果使用for循環(huán)遍歷，一次發(fā)送http請求，所有請求都是串行，如果一個請求需要1秒，25個請求就需要25秒。
解法方案：思路 ：異步請求，多線程發(fā)送http。
具體方案：
方案一：自定義線程池，定義請求方法，使用for循環(huán)講請求添加到線程詞。
方案二（推薦）：異步請求注解,使用springboot的@Async注解，完美解決，編碼量少。
方案三：并行流（Parallel Stream）。
以下是方案三詳解：
并行流（Parallel Stream）深度解析
一、并行流是什么？
并行流是Java 8引入的一個特性，它允許你將流操作并行執(zhí)行，利用多核處理器的優(yōu)勢來加速處理。
java
// 順序流 vs 并行流
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
// 1. 順序流（默認）
int sumSequential = numbers.stream()
.reduce(0, Integer::sum);
// 2. 并行流
int sumParallel = numbers.parallelStream()
.reduce(0, Integer::sum);
// 3. 將順序流轉為并行流
int sumParallel2 = numbers.stream()
.parallel()
.reduce(0, Integer::sum);二、并行流的核心原理
1. 底層框架：Fork/Join框架
java
// 并行流背后是這個框架
ForkJoinPool commonPool = ForkJoinPool.commonPool();
System.out.println("并行度: " + commonPool.getParallelism()); // 通常是CPU核心數(shù)-12. 工作竊取算法（Work-Stealing）
text
線程A：│任務1│任務2│任務3│任務4│
線程B：│任務5│任務6│ │ ← 空閑
↓
線程B從線程A的隊尾"偷取"任務3來執(zhí)行3. 數(shù)據(jù)拆分策略
java
// 并行流自動將數(shù)據(jù)拆分為多個子任務 Spliterator<T> spliterator = list.spliterator(); // 查看分割特性 int characteristics = spliterator.characteristics(); // SIZED: 知道確切大小 // SUBSIZED: 分割后的大小也確切 // ORDERED: 有順序要求
三、基本使用模式
1. 創(chuàng)建并行流
java
// 方式1：從集合創(chuàng)建
List<String> list = Arrays.asList("a", "b", "c");
Stream<String> parallelStream = list.parallelStream();
// 方式2：將順序流轉為并行流
Stream<String> parallelStream2 = list.stream().parallel();
// 方式3：從數(shù)組創(chuàng)建
String[] array = {"a", "b", "c"};
Stream<String> parallelStream3 = Arrays.stream(array).parallel();2. 常用操作示例
映射操作（map）
java
List<String> words = Arrays.asList("hello", "world", "java", "stream");
// 順序處理
List<String> upperCaseSequential = words.stream()
.map(String::toUpperCase)
.collect(Collectors.toList());
// 并行處理
List<String> upperCaseParallel = words.parallelStream()
.map(String::toUpperCase)
.collect(Collectors.toList());過濾操作（filter）
java
List<Integer> numbers = IntStream.range(1, 1000000).boxed().collect(Collectors.toList());
// 并行過濾偶數(shù)
List<Integer> evenNumbers = numbers.parallelStream()
.filter(n -> n % 2 == 0)
.collect(Collectors.toList());歸約操作（reduce）
java
// 計算1到1000000的和（并行版本更快）
long sum = LongStream.rangeClosed(1, 1000000)
.parallel()
.reduce(0, Long::sum);四、你的代碼中并行流的應用
java
private List<Map<String, String>> getInterlockResultWen13or24Level1PointValueParallel(
Map<String, String> map,
List<IntermediateTable> list
) {
// 1. 建立查找表（提升效率）
Map<String, String> codeToTagNumber = list.stream()
.collect(Collectors.toMap(
IntermediateTable::getInterlockWen13or24Level1Point,
IntermediateTable::getTagNumber,
(v1, v2) -> v1 // 如果有重復key，取第一個
));
// 2. 使用并行流處理
return map.entrySet()
.parallelStream() // ← 關鍵！開啟并行
.map(entry -> {
String code = entry.getKey();
String name = entry.getValue();
String tagNumber = codeToTagNumber.get(code);
// 3. 執(zhí)行IO操作（HTTP請求）
String value = tagNumber != null ? getSupOSValue(tagNumber) : "0";
// 4. 構建結果
Map<String, String> resMap = new LinkedHashMap<>();
resMap.put("name", name);
resMap.put("value", value);
return resMap;
})
.collect(Collectors.toList()); // 收集結果
}五、性能對比：串行 vs 并行
java
public class ParallelStreamBenchmark {
public static void main(String[] args) {
// 模擬24次HTTP請求（用睡眠模擬）
List<Integer> requestIds = IntStream.range(1, 25).boxed().collect(Collectors.toList());
// 串行執(zhí)行
long start = System.currentTimeMillis();
List<String> sequentialResults = requestIds.stream()
.map(id -> mockHttpRequest(id))
.collect(Collectors.toList());
long sequentialTime = System.currentTimeMillis() - start;
// 并行執(zhí)行
start = System.currentTimeMillis();
List<String> parallelResults = requestIds.parallelStream()
.map(id -> mockHttpRequest(id))
.collect(Collectors.toList());
long parallelTime = System.currentTimeMillis() - start;
System.out.println("串行耗時: " + sequentialTime + "ms");
System.out.println("并行耗時: " + parallelTime + "ms");
System.out.println("加速比: " + (double)sequentialTime/parallelTime);
}
private static String mockHttpRequest(Integer id) {
try {
// 模擬HTTP請求延遲（100-500ms隨機）
Thread.sleep(100 + (long)(Math.random() * 400));
return "Response_" + id;
} catch (InterruptedException e) {
return "Error";
}
}
}六、注意事項和最佳實踐
1. 什么時候使用并行流？
java
// ? 適合的場景 - 數(shù)據(jù)量大（>1000個元素） - 每個元素的處理是獨立的 - 處理操作是CPU密集型或IO密集型 - 沒有順序要求（或者可以使用unordered()放棄順序） // ? 不適合的場景 - 數(shù)據(jù)量?。?lt;100個元素） - 處理操作簡單（如簡單的數(shù)學運算） - 需要嚴格保持順序 - 操作有共享狀態(tài)（線程不安全）
2. 線程安全問題
java
// ? 錯誤示例：共享可變狀態(tài)
List<Integer> sharedList = new ArrayList<>();
numbers.parallelStream()
.forEach(sharedList::add); // 并發(fā)修改異常！
// ? 正確做法：使用線程安全收集器
List<Integer> safeList = numbers.parallelStream()
.collect(Collectors.toList());
// ? 或者使用同步集合
List<Integer> synchronizedList = Collections.synchronizedList(new ArrayList<>());
numbers.parallelStream()
.forEach(synchronizedList::add);3. 性能優(yōu)化技巧
java
// 1. 使用基礎類型流避免裝箱拆箱
IntStream.range(1, 1000000)
.parallel()
.sum();
// 2. 使用unordered()放棄順序約束（可能提升性能）
list.parallelStream()
.unordered()
.filter(...)
.collect(Collectors.toList());
// 3. 合并操作（combiner）必須是可結合的
// reduce操作需要：a ⊕ b ⊕ c = a ⊕ (b ⊕ c)4. 控制并行度
java
// 方法1：修改系統(tǒng)屬性
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");
// 方法2：使用自定義ForkJoinPool
ForkJoinPool customPool = new ForkJoinPool(4);
List<String> results = customPool.submit(() ->
list.parallelStream()
.map(item -> process(item))
.collect(Collectors.toList())
).get();七、在你的具體場景中的應用建議
java
private List<Map<String, String>> getInterlockResultWen13or24Level1PointValueOptimized(
Map<String, String> map,
List<IntermediateTable> list
) {
// 1. 建立索引
Map<String, String> codeToTag = list.stream()
.collect(Collectors.toMap(
IntermediateTable::getInterlockWen13or24Level1Point,
IntermediateTable::getTagNumber,
(v1, v2) -> v1
));
// 2. 預計算需要請求的tagNumbers
List<String> tagNumbers = map.keySet().stream()
.map(codeToTag::get)
.filter(Objects::nonNull)
.distinct()
.collect(Collectors.toList());
// 3. 并行獲取所有值
Map<String, String> valueCache = tagNumbers.parallelStream()
.collect(Collectors.toMap(
tag -> tag,
this::getSupOSValue // 這是你的HTTP請求方法
));
// 4. 構建結果
return map.entrySet().stream()
.map(entry -> {
String code = entry.getKey();
String tag = codeToTag.get(code);
String value = tag != null ? valueCache.getOrDefault(tag, "0") : "0";
Map<String, String> result = new LinkedHashMap<>();
result.put("name", entry.getValue());
result.put("value", value);
return result;
})
.collect(Collectors.toList());
}八、調(diào)試和監(jiān)控
java
// 1. 查看并行流使用的線程
List<String> results = list.parallelStream()
.peek(item -> System.out.println(Thread.currentThread().getName()))
.map(...)
.collect(Collectors.toList());
// 2. 監(jiān)控執(zhí)行時間
long start = System.nanoTime();
List<String> results = list.parallelStream()
.map(...)
.collect(Collectors.toList());
long duration = System.nanoTime() - start;
// 3. 使用JMH進行基準測試（推薦）
// 添加依賴：org.openjdk.jmh九、常見陷阱
java
// 陷阱1：副作用操作
List<String> list = Arrays.asList("a", "b", "c");
StringBuilder builder = new StringBuilder();
list.parallelStream().forEach(builder::append); // ? 線程不安全
// 陷阱2：有狀態(tài)lambda
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> results = new ArrayList<>();
numbers.parallelStream()
.filter(n -> {
results.add(n); // ? 在lambda中修改外部狀態(tài)
return n % 2 == 0;
})
.collect(Collectors.toList());
// 陷阱3：順序依賴
Optional<Integer> first = numbers.parallelStream()
.filter(n -> n > 3)
.findFirst(); // ?? 并行下findFirst可能變慢
總結
到此這篇關于java并行流處理的文章就介紹到這了,更多相關java并行流處理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Hadoop集成Spring的使用詳細教程(快速入門大數(shù)據(jù))
這篇文章主要介紹了Hadoop集成Spring的使用詳細教程(快速入門大數(shù)據(jù)),本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-01-01
這篇文章主要介紹了Maven工程pom中如何定義jdk版本，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
springboot項目攔截器重定向循環(huán)問題的解決
這篇文章主要介紹了springboot項目攔截器重定向循環(huán)問題的解決，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-09-09
java 實現(xiàn)隨機數(shù)組輸出及求和實例詳解
這篇文章主要介紹了java 實現(xiàn)隨機數(shù)組輸出及求和實例詳解的相關資料,需要的朋友可以參考下2016-11-11
這篇文章主要介紹了Java設計模式之工廠方法模式,較為詳細的分析了Java工廠模式的功能、定義并給出了實例代碼加以總結分析,需要的朋友可以參考下2015-11-11
這篇文章主要介紹了java實現(xiàn)樹形查詢方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-07-07
Java實現(xiàn)DFA算法對敏感詞、廣告詞過濾功能示例
本篇文章主要介紹了Java實現(xiàn)DFA算法對敏感詞、廣告詞過濾功能示例，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-11-11
Oracle + Mybatis實現(xiàn)批量插入、更新和刪除示例代碼
利用MyBatis動態(tài)SQL的特性,我們可以做一些批量的操作，下面這篇文章主要給大家介紹了關于Oracle + Mybatis實現(xiàn)批量插入、更新和刪除的相關資料，文中通過示例代碼介紹的非常詳細，需要的朋友可以參考借鑒，下面來一起看看吧。2018-01-01
本文主要介紹了@Accessors和@Builder使用,區(qū)別與坑,對使用這兩個注解的有一定的幫助,具有一定的參考價值,感興趣的可以了解一下2023-08-08
JAVA面試題之Forward與Redirect的區(qū)別詳解
這篇文章主要給大家介紹了在JAVA面試中可能遇到會遇到的一道題，就是java中Forward與Redirect兩者之前的區(qū)別，文中介紹的非常詳細，對大家具有一定參考學習價值，需要的朋友們下面來一起看看吧。2017-05-05