Java8中的并行流處理-順序與并行流處理的性能

并行處理現(xiàn)在隨處可見(jiàn)。由于 CPU 核數(shù)量的增加以及較低的硬件成本使得集群系統(tǒng)更加便宜，并行處理似乎成為下一個(gè)大趨勢(shì)。

Java8通過(guò)新的 stream API 和在集合和數(shù)組上創(chuàng)建并行處理的簡(jiǎn)化來(lái)關(guān)注這個(gè)問(wèn)題。讓我們看看這是怎么運(yùn)作的。

假設(shè) myList 是一個(gè)整數(shù)列表，包含500,000個(gè)整數(shù)值。在 java 8 時(shí)代之前，對(duì)這個(gè)整數(shù)值求和的方法是使用 for 每個(gè)循環(huán)。

for (int i :myList) result+=i;

自從 java 8 之后我們可以使用 stream 來(lái)做同樣的事情

myList.stream().sum();

并行化這個(gè)過(guò)程非常簡(jiǎn)單，如果我們?nèi)匀挥幸粋€(gè)流，我們只需要使用關(guān)鍵字 parallelStream() 來(lái)代替 stream 或者 parallel()。

那么
myList.parallelStream().sum()
會(huì)有用的。因此，很容易將計(jì)算擴(kuò)展到線程和可用的 CPU 核心。但是我們知道多線程和并行處理的開(kāi)銷(xiāo)是昂貴的。問(wèn)題是什么時(shí)候使用并行流，以及什么時(shí)候串行流更好地與性能相關(guān)。

首先讓我們來(lái)看看幕后發(fā)生了什么。并行流使用 Fork/Join Framework 進(jìn)行處理。這意味著 stream-source 將被拆分并交給 fork/join-pool workers 執(zhí)行。

但是在這里我們首先要考慮的是，并非所有的流源都像其他流源一樣可以分割?？紤]一個(gè) ArrayList，它的內(nèi)部數(shù)據(jù)表示基于一個(gè)數(shù)組。拆分這樣的源非常容易，因?yàn)榭梢杂?jì)算中間元素的索引并拆分?jǐn)?shù)組。

如果我們有一個(gè) LinkedList，那么分割數(shù)據(jù)元素就會(huì)更加復(fù)雜。實(shí)現(xiàn)必須遍歷第一個(gè)條目中的所有元素，以找到可以進(jìn)行拆分的元素。例如，LinkedLists 對(duì)于并行流的性能就很差。

這是我們可以保留的關(guān)于并行流性能的第一個(gè)事實(shí)。

S – 源集合必須可以有效拆分

拆分集合、管理 fork 和 join 任務(wù)、對(duì)象創(chuàng)建和垃圾收集也是一個(gè)算法開(kāi)銷(xiāo)。當(dāng)且僅當(dāng)在CPU核心上可簡(jiǎn)單完成或者集合足夠大時(shí)，才值得這樣做。

一個(gè)不好的例子是計(jì)算5個(gè)整數(shù)值的最大值。

IntStream.rangeClosed(1, 5).reduce( Math::max).getAsInt();

系統(tǒng)為 fork/join 準(zhǔn)備和處理數(shù)據(jù)的開(kāi)銷(xiāo)非常大，以至于串行流在此場(chǎng)景中要快得多。Math.Max 函數(shù)在這里的 CPU 開(kāi)銷(xiāo)并不是很高，而且數(shù)據(jù)元素很少。

但當(dāng)每個(gè)元素執(zhí)行的函數(shù)更加復(fù)雜一些——變得「更加 CPU 密集型」時(shí)，那它就越來(lái)越值得這么做了。比如計(jì)算每個(gè)元素的正弦值而不是最大值。

在編寫(xiě)一個(gè)國(guó)際象棋游戲時(shí)，計(jì)算每一步棋的走法也是一個(gè)這種類(lèi)型的例子。 大量的計(jì)算可以并行完成。并且下一步棋的走法有很多種可能性。

這種情況就非常適合使用并行處理。

這就是我們可以保留的關(guān)于并行流性能的第二個(gè)事實(shí)：

N*Q —— 「元素個(gè)數(shù) * 每個(gè)元素的操作成本」因子 應(yīng)該要很大

但這也意味著反過(guò)來(lái)，當(dāng)每個(gè)元素的操作成本較高時(shí)，集合應(yīng)該更小。
或者，當(dāng)每個(gè)元素的操作不是CPU密集型操作時(shí)，我們需要一個(gè)包含許多元素的非常大的集合，以便并行流的使用變得有價(jià)值。

這就直接取決于我們可以保留的第三個(gè)事實(shí)：

C —— CPU核心數(shù) —— 越多越好 必須大于1

在單核機(jī)器上，由于一些管理上的開(kāi)銷(xiāo)導(dǎo)致并行流的性能總是比串行流的差。這就和有多個(gè)項(xiàng)目負(fù)責(zé)人和一個(gè)做事的人的公司是一樣的。

越多越好 ——不幸的是，這對(duì)于現(xiàn)實(shí)生活中大多數(shù)情況都不適用，比如對(duì)于一個(gè)特別小的集合，多個(gè) CPU 核心啟動(dòng)，然后發(fā)現(xiàn)無(wú)事可做——也許他們之前處于節(jié)能模式。

另外，決定是否使用并行流對(duì)于每一個(gè)元素執(zhí)行的函數(shù)也有一些要求。這與其說(shuō)是性能問(wèn)題，不如說(shuō)是并行流能否按預(yù)期工作的問(wèn)題。
函數(shù)必須滿足以下要求：

• 獨(dú)立。這就意味著對(duì)于每個(gè)元素的計(jì)算一定不能依賴或者影響任何其他元素。

• 不干預(yù)。 這就意味著函數(shù)在處理過(guò)程中不會(huì)修改底層數(shù)據(jù)源。

• 無(wú)狀態(tài)。

下面是一個(gè)在并行流中使用有狀態(tài) lambda 函數(shù)的例子。這個(gè)例子是從 Java 的 JDK API 里抽取出來(lái)的對(duì) distinct() 的簡(jiǎn)易實(shí)現(xiàn)。
Set seen = Collections.synchronizedSet(new HashSet()); stream.parallel().map(e -> { if (seen.add(e)) return 0; else return e; })...

這就引出了我們可以保留的第四個(gè)事實(shí)：

F —— 每個(gè)元素的操作函數(shù)必須是獨(dú)立的

綜上所述，即：

那還有別的不應(yīng)該使用并行流的情況嗎？有，當(dāng)然有。

始終考慮每個(gè)元素的函數(shù)正在做什么，以及這是否適合用在并行流里。當(dāng)你的函數(shù)在調(diào)用一些同步功能時(shí)，你可能不會(huì)因?yàn)椴⑿谢鞫@益，因?yàn)椴⑿辛魍ǔ?huì)在這個(gè)同步上阻塞。

當(dāng)你調(diào)用阻塞的 I/O 操作，也會(huì)發(fā)生同樣的問(wèn)題。

就這一點(diǎn)來(lái)說(shuō)，使用基于 I/O 的源作為流也是眾所周知的性能不好，因?yàn)閿?shù)據(jù)是按照順序讀取的，所以這樣的源很難拆分。