前言

parallelStream() 是 Java 8 Stream API 提供的 并行流（Parallel Stream）

它可以讓：

• 一個(gè)集合的處理邏輯
• 自動(dòng)使用多線程并行執(zhí)行

而不是單線程。

一、最簡單理解

普通 stream：

list.stream()

是 單線程

例如：

1個(gè)線程處理20萬數(shù)據(jù)

parallelStream：

list.parallelStream()

是 多線程并行處理

例如：

8核CPU
可能8個(gè)線程同時(shí)處理

二、舉個(gè)最簡單例子

普通 stream ：

list.stream()
    .map(x -> doSomething(x))
    .collect(Collectors.toList());

執(zhí)行過程：

第1條
第2條
第3條
...

按順序一個(gè)個(gè)執(zhí)行。

parallelStream ：

list.parallelStream()
    .map(x -> doSomething(x))
    .collect(Collectors.toList());

執(zhí)行過程：

線程1 -> 第1萬條
線程2 -> 第2萬條
線程3 -> 第3萬條
...

同時(shí)執(zhí)行。

三、parallelStream底層是什么？

底層使用： ForkJoinPool

默認(rèn)：

CPU核心數(shù) - 1

例如：

8核CPU：

默認(rèn)7個(gè)工作線程

四、parallelStream什么時(shí)候適合？

適合：CPU密集型任務(wù)

例如：

• 大量計(jì)算
• 加密
• 哈希
• 復(fù)雜規(guī)則計(jì)算
• 圖片處理
• 金額計(jì)算
• 風(fēng)控計(jì)算

例如：

list.parallelStream()
    .map(this::calculate)

計(jì)算 很耗CPU。這時(shí)收益巨大。

五、什么時(shí)候不適合？

IO密集型

例如：

數(shù)據(jù)庫
Feign
HTTP
Redis
文件

例如：

list.parallelStream()
    .forEach(x -> feign.call());

通常會(huì)更慢，甚至打爆數(shù)據(jù)庫/接口。

六、為什么IO場景不適合？

因?yàn)?，parallelStream 默認(rèn)線程數(shù)：

CPU核數(shù)

而IO：

大量時(shí)間在等待

CPU其實(shí)沒干活。

比如：

線程1 等數(shù)據(jù)庫
線程2 等網(wǎng)絡(luò)
線程3 等Redis

CPU閑著。

所以：

parallelStream 對IO提升有限。

七、parallelStream的優(yōu)勢

1. 開發(fā)簡單

不用：

線程池
Future
CountDownLatch
CompletableFuture

一行搞定。

2. 自動(dòng)線程調(diào)度

自動(dòng)：

• 分片
• 分任務(wù)
• 合并結(jié)果

3. CPU利用率高

例如：

8核CPU：

單線程只用了1核

parallelStream：

8核一起跑

八、parallelStream的缺點(diǎn)

很多人亂用。

缺點(diǎn)1：線程不受控

默認(rèn)：

ForkJoinPool.commonPool()

全局共享線程池。

可能：

影響整個(gè)系統(tǒng)

缺點(diǎn)2：不適合數(shù)據(jù)庫/Feign

例如：

parallelStream()
    .forEach(x -> mapper.insert(x));

危險(xiǎn)。

可能 瞬間幾千SQL。

缺點(diǎn)3：線程安全問題

例如：

你這樣寫：

List<String> result = new ArrayList<>();

list.parallelStream().forEach(x -> {
    result.add(x);
});

錯(cuò)誤！

因?yàn)椋?/p>

ArrayList線程不安全

可能：

• 數(shù)據(jù)丟失
• 數(shù)組越界
• ConcurrentModificationException

正確寫法

用：

collect(Collectors.toList())

因?yàn)?collect 內(nèi)部處理了并發(fā)。

九、parallelStream為什么有時(shí)反而更慢？

因?yàn)椋?strong>線程切換有成本

例如：

1+1

這種極小任務(wù)。

還沒切線程快。

所以：數(shù)據(jù)量小不要用

經(jīng)驗(yàn)：

十、parallelStream不保證順序

例如：

list.parallelStream()
    .forEach(System.out::println);

輸出：

亂序

如果需要順序：

forEachOrdered()

但：性能會(huì)下降。

十一、真正高性能方案（推薦）

很多大系統(tǒng)：

不會(huì)直接用：

parallelStream

而是：

自定義線程池 + CompletableFuture

因?yàn)椋?/p>

• 可控
• 可監(jiān)控
• 可限流
• 可隔離

例如：

ExecutorService pool = Executors.newFixedThreadPool(8);

CompletableFuture.supplyAsync(() -> {
    return calculate();
}, pool);

十二、parallelStream vs 線程池

到此這篇關(guān)于Java parallelStream并行流使用的文章就介紹到這了,更多相關(guān)Java parallelStream并行流內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論