Java 8 API添加了一個(gè)新的抽象稱(chēng)為流Stream，可以讓你以一種聲明的方式處理數(shù)據(jù)。

Stream 使用一種類(lèi)似用 SQL 語(yǔ)句從數(shù)據(jù)庫(kù)查詢(xún)數(shù)據(jù)的直觀方式來(lái)提供一種對(duì) Java 集合運(yùn)算和表達(dá)的高階抽象。

Stream API可以極大提高Java程序員的生產(chǎn)力，讓程序員寫(xiě)出高效率、干凈、簡(jiǎn)潔的代碼。

本文會(huì)對(duì)Stream的實(shí)現(xiàn)原理進(jìn)行剖析。

Stream的組成與特點(diǎn)

Stream（流）是一個(gè)來(lái)自數(shù)據(jù)源的元素隊(duì)列并支持聚合操作：

• 元素是特定類(lèi)型的對(duì)象，形成一個(gè)隊(duì)列。?Java中的Stream并_不會(huì)_向集合那樣存儲(chǔ)和管理元素，而是按需計(jì)算

• 數(shù)據(jù)源流的來(lái)源可以是集合Collection、數(shù)組Array、I/O channel， 產(chǎn)生器generator?等

• 聚合操作類(lèi)似SQL語(yǔ)句一樣的操作， 比如filter,?map,?reduce,?find,?match,?sorted等

和以前的Collection操作不同， Stream操作還有兩個(gè)基礎(chǔ)的特征：

• Pipelining: 中間操作都會(huì)返回流對(duì)象本身。這樣多個(gè)操作可以串聯(lián)成一個(gè)管道， 如同流式風(fēng)格（fluent style）。這樣做可以對(duì)操作進(jìn)行優(yōu)化， 比如延遲執(zhí)行(laziness evaluation)和短路(?short-circuiting)

• 內(nèi)部迭代：以前對(duì)集合遍歷都是通過(guò)Iterator或者For-Each的方式, 顯式的在集合外部進(jìn)行迭代， 這叫做外部迭代。?Stream提供了內(nèi)部迭代的方式， 通過(guò)訪(fǎng)問(wèn)者模式 (Visitor)實(shí)現(xiàn)。

和迭代器又不同的是，Stream?可以并行化操作，迭代器只能命令式地、串行化操作。顧名思義，當(dāng)使用串行方式去遍歷時(shí)，每個(gè)?item?讀完后再讀下一個(gè) item。而使用并行去遍歷時(shí)，數(shù)據(jù)會(huì)被分成多個(gè)段，其中每一個(gè)都在不同的線(xiàn)程中處理，然后將結(jié)果一起輸出。

Stream?的并行操作依賴(lài)于?Java7?中引入的?Fork/Join?框架（JSR166y）來(lái)拆分任務(wù)和加速處理過(guò)程。Java?的并行 API 演變歷程基本如下：

1.0-1.4 中的 java.lang.Thread

5.0 中的 java.util.concurrent

6.0 中的 Phasers 等

7.0 中的 Fork/Join 框架

8.0 中的 Lambda

Stream具有平行處理能力，處理的過(guò)程會(huì)分而治之，也就是將一個(gè)大任務(wù)切分成多個(gè)小任務(wù)，這表示每個(gè)任務(wù)都是一個(gè)操作：

List?numbers?=?Arrays.asList(1,?2,?3,?4,?5,?6,?7,?8,?9);
numbers.parallelStream()
???????.forEach(out::println);?

可以看到一行簡(jiǎn)單的代碼就幫我們實(shí)現(xiàn)了并行輸出集合中元素的功能，但是由于并行執(zhí)行的順序是不可控的所以每次執(zhí)行的結(jié)果不一定相同。

如果非得相同可以使用forEachOrdered方法執(zhí)行終止操作：

List?numbers?=?Arrays.asList(1,?2,?3,?4,?5,?6,?7,?8,?9);
numbers.parallelStream()
???????.forEachOrdered(out::println);??

這里有一個(gè)疑問(wèn)，如果結(jié)果需要有序，是否和我們的并行執(zhí)行的初衷相悖？是的，這個(gè)場(chǎng)景下明顯無(wú)需使用并行流，直接用串行流執(zhí)行即可， 否則性能可能更差，因?yàn)樽詈笥謴?qiáng)行將所有并行結(jié)果進(jìn)行了排序。

OK，下面我們先介紹一下Stream接口的相關(guān)知識(shí)。

BaseStream接口

Stream的父接口是BaseStream，后者是所有流實(shí)現(xiàn)的頂層接口，定義如下：

public?interface?BaseStream<T,?S?extends?BaseStream<T,?S>>
????????extends?AutoCloseable?{
????Iterator?iterator();

????Spliterator?spliterator();

????boolean?isParallel();

????S?sequential();

????S?parallel();

????S?unordered();

????S?onClose(Runnable?closeHandler);

????void?close();
}

其中，T為流中元素的類(lèi)型，S為一個(gè)BaseStream的實(shí)現(xiàn)類(lèi)，它里面的元素也是T并且S同樣是自己：

S extends BaseStream

是不是有點(diǎn)暈？

其實(shí)很好理解，我們看一下接口中對(duì)S的使用就知道了：如sequential()、parallel()這兩個(gè)方法，它們都返回了S實(shí)例，也就是說(shuō)它們分別支持對(duì)當(dāng)前流進(jìn)行串行或者并行的操作，并返回「改變」后的流對(duì)象。

如果是并行一定涉及到對(duì)當(dāng)前流的拆分，即將一個(gè)流拆分成多個(gè)子流，子流肯定和父流的類(lèi)型是一致的。子流可以繼續(xù)拆分子流，一直拆分下去…

也就是說(shuō)這里的S是BaseStream的一個(gè)實(shí)現(xiàn)類(lèi)，它同樣是一個(gè)流，比如Stream、IntStream、LongStream等。

Stream接口

再來(lái)看一下Stream的接口聲明：

public?interface?Stream?extends?BaseStream>?

參考上面的解釋這里不難理解：即Stream可以繼續(xù)拆分為Stream，我們可以通過(guò)它的一些方法來(lái)證實(shí)：

Stream?filter(Predicatesuper?T>?predicate);
?Stream?map(Functionsuper?T,???extends?R>?mapper);
?Stream?flatMap(Functionsuper?T,???extends?Stream>?mapper);
Stream?sorted();
Stream?peek(Consumersuper?T>?action);
Stream?limit(long?maxSize);
Stream?skip(long?n);
...

這些都是操作流的中間操作，它們的返回結(jié)果必須是流對(duì)象本身。

關(guān)閉流操作

BaseStream 實(shí)現(xiàn)了?AutoCloseable?接口，也就是?close()?方法會(huì)在流關(guān)閉時(shí)被調(diào)用。同時(shí)，BaseStream?中還給我們提供了onClose()方法：

/**?*?Returns?an?equivalent?stream?with?an?additional?close?handler.?Close?*?handlers?are?run?when?the?{@link?#close()}?method?*?is?called?on?the?stream,?and?are?executed?in?the?order?they?were?*?added.?All?close?handlers?are?run,?even?if?earlier?close?handlers?throw?*?exceptions.?If?any?close?handler?throws?an?exception,?the?first?*?exception?thrown?will?be?relayed?to?the?caller?of?{@code?close()},?with?*?any?remaining?exceptions?added?to?that?exception?as?suppressed?exceptions?*?(unless?one?of?the?remaining?exceptions?is?the?same?exception?as?the?*?first?exception,?since?an?exception?cannot?suppress?itself.)?May?*?return?itself.?*?*?
This?is?an?intermediate?*?operation.?*?*?@param?closeHandler?A?task?to?execute?when?the?stream?is?closed?*?@return?a?stream?with?a?handler?that?is?run?if?the?stream?is?closed?*/
S?onClose(Runnable?closeHandler);

當(dāng)AutoCloseable的close()接口被調(diào)用的時(shí)候會(huì)觸發(fā)調(diào)用流對(duì)象的onClose()方法，但有幾點(diǎn)需要注意：

• onClose()?方法會(huì)返回流對(duì)象本身，也就是說(shuō)可以對(duì)改對(duì)象進(jìn)行多次調(diào)用

• 如果調(diào)用了多個(gè)onClose()?方法，它會(huì)按照調(diào)用的順序觸發(fā)，但是如果某個(gè)方法有異常則只會(huì)向上拋出第一個(gè)異常

• 前一個(gè)?onClose()?方法拋出了異常不會(huì)影響后續(xù)?onClose()?方法的使用

• 如果多個(gè)?onClose()?方法都拋出異常，只展示第一個(gè)異常的堆棧，而其他異常會(huì)被壓縮，只展示部分信息

并行流和串行流

BaseStream接口中分別提供了并行流和串行流兩個(gè)方法，這兩個(gè)方法可以任意調(diào)用若干次，也可以混合調(diào)用，但最終只會(huì)以最后一次方法調(diào)用的返回結(jié)果為準(zhǔn)。

參考parallel()方法的說(shuō)明：

Returns an equivalent stream that is parallel. May return

itself, either because the stream was already parallel, or because

the underlying stream state was modified to be parallel.

所以多次調(diào)用同樣的方法并不會(huì)生成新的流，而是直接復(fù)用當(dāng)前的流對(duì)象。

下面的例子里以最后一次調(diào)用parallel()為準(zhǔn)，最終是并行地計(jì)算sum：

stream.parallel()
???.filter(...)
???.sequential()
???.map(...)
???.parallel()
???.sum();

ParallelStream背后的男人：ForkJoinPool

ForkJoin框架是從JDK7中新特性，它同ThreadPoolExecutor一樣，也實(shí)現(xiàn)了Executor和ExecutorService 接口。它使用了一個(gè)「無(wú)限隊(duì)列」來(lái)保存需要執(zhí)行的任務(wù)，而線(xiàn)程的數(shù)量則是通過(guò)構(gòu)造函數(shù)傳入，?如果沒(méi)有向構(gòu)造函數(shù)中傳入希望的線(xiàn)程數(shù)量，那么當(dāng)前計(jì)算機(jī)可用的CPU數(shù)量會(huì)被設(shè)置為線(xiàn)程數(shù)量作為默認(rèn)值。

ForkJoinPool主要用來(lái)使用分治法(Divide-and-Conquer Algorithm) 來(lái)解決問(wèn)題，典型的應(yīng)用比如_快速排序算法_。這里的要點(diǎn)在于，F(xiàn)orkJoinPool需要使用相對(duì)少的線(xiàn)程來(lái)處理大量的任務(wù)。比如要對(duì)1000萬(wàn)個(gè)數(shù)據(jù)進(jìn)行排序，那么會(huì)將這個(gè)任務(wù)分割成兩個(gè)500 萬(wàn)的排序任務(wù)和一個(gè)針對(duì)這兩組500萬(wàn)數(shù)據(jù)的合并任務(wù)。

以此類(lèi)推，對(duì)于500萬(wàn)的數(shù)據(jù)也會(huì)做出同樣的分割處理，到最后會(huì)設(shè)置一個(gè)閾值來(lái)規(guī)定當(dāng)數(shù)據(jù)規(guī)模到多少時(shí)，停止這樣的分割處理。比如，當(dāng)元素的數(shù)量小于10時(shí)，會(huì)停止分割，轉(zhuǎn)而使用插入排序?qū)λ鼈冞M(jìn)行排序。那么到最后，所有的任務(wù)加起來(lái)會(huì)有大概2000000+個(gè)。

問(wèn)題的關(guān)鍵在于，對(duì)于一個(gè)任務(wù)而言，只有當(dāng)它所有的子任務(wù)完成之后，它才能夠被執(zhí)行，想象一下歸并排序的過(guò)程。

所以當(dāng)使用ThreadPoolExecutor時(shí)，使用分治法會(huì)存在問(wèn)題，因?yàn)門(mén)hreadPoolExecutor中的線(xiàn)程無(wú)法向 任務(wù)隊(duì)列中再添加一個(gè)任務(wù)并且在等待該任務(wù)完成之后再繼續(xù)執(zhí)行。而使用ForkJoinPool時(shí)，就能夠讓其中的線(xiàn)程創(chuàng)建新的任務(wù)，并掛起當(dāng)前的任務(wù)，此時(shí)線(xiàn)程就能夠從隊(duì)列中選擇子任務(wù)執(zhí)行。

那么使用ThreadPoolExecutor或者ForkJoinPool，會(huì)有什么性能的差異呢？

首先，使用ForkJoinPool能夠使用數(shù)量有限的線(xiàn)程來(lái)完成非常多的具有「父子關(guān)系」的任務(wù)，比如使用4個(gè)線(xiàn)程來(lái)完成超過(guò)200萬(wàn)個(gè)任務(wù)。使用ThreadPoolExecutor 時(shí)，是不可能完成的，因?yàn)門(mén)hreadPoolExecutor中的Thread無(wú)法選擇優(yōu)先執(zhí)行子任務(wù)，需要完成200萬(wàn)個(gè)具有父子關(guān)系的任務(wù)時(shí)，也需要200萬(wàn)個(gè)線(xiàn)程，顯然這是不可行的。

Work Stealing原理：

1. 每個(gè)工作線(xiàn)程都有自己的工作隊(duì)列WorkQueue；
2. 這是一個(gè)雙端隊(duì)列dequeue，它是線(xiàn)程私有的；
3. ForkJoinTask中fork的子任務(wù)，將放入運(yùn)行該任務(wù)的工作線(xiàn)程的隊(duì)頭，工作線(xiàn)程將以L(fǎng)IFO的順序來(lái)處理工作隊(duì)列中的任務(wù)，即堆棧的方式；
4. 為了最大化地利用CPU，空閑的線(xiàn)程將從其它線(xiàn)程的隊(duì)列中「竊取」任務(wù)來(lái)執(zhí)行
5. 但是是從工作隊(duì)列的尾部竊取任務(wù)，以減少和隊(duì)列所屬線(xiàn)程之間的競(jìng)爭(zhēng)；
6. 雙端隊(duì)列的操作：push()/pop()僅在其所有者工作線(xiàn)程中調(diào)用，poll()是由其它線(xiàn)程竊取任務(wù)時(shí)調(diào)用的；
7. 當(dāng)只剩下最后一個(gè)任務(wù)時(shí)，還是會(huì)存在競(jìng)爭(zhēng)，是通過(guò)CAS來(lái)實(shí)現(xiàn)的；

用ForkJoinPool的眼光來(lái)看ParallelStream

Java 8為ForkJoinPool添加了一個(gè)通用線(xiàn)程池，這個(gè)線(xiàn)程池用來(lái)處理那些沒(méi)有被顯式提交到任何線(xiàn)程池的任務(wù)。它是ForkJoinPool類(lèi)型上的一個(gè)靜態(tài)元素，它擁有的默認(rèn)線(xiàn)程數(shù)量等于運(yùn)行計(jì)算機(jī)上的CPU數(shù)量。當(dāng)調(diào)用Arrays 類(lèi)上添加的新方法時(shí)，自動(dòng)并行化就會(huì)發(fā)生。比如用來(lái)排序一個(gè)數(shù)組的并行快速排序，用來(lái)對(duì)一個(gè)數(shù)組中的元素進(jìn)行并行遍歷。自動(dòng)并行化也被運(yùn)用在Java 8新添加的Stream API中。

比如下面的代碼用來(lái)遍歷列表中的元素并執(zhí)行需要的操作：

List?userInfoList?=
????????DaoContainers.getUserInfoDAO().queryAllByList(new?UserInfoModel());
userInfoList.parallelStream().forEach(RedisUserApi::setUserIdUserInfo);

對(duì)于列表中的元素的操作都會(huì)以并行的方式執(zhí)行。forEach方法會(huì)為每個(gè)元素的計(jì)算操作創(chuàng)建一個(gè)任務(wù)，該任務(wù)會(huì)被前文中提到的ForkJoinPool中的commonPool處理。以上的并行計(jì)算邏輯當(dāng)然也可以使用ThreadPoolExecutor完成，但是就代碼的可讀性和代碼量而言，使用ForkJoinPool明顯更勝一籌。

對(duì)于ForkJoinPool通用線(xiàn)程池的線(xiàn)程數(shù)量，通常使用默認(rèn)值就可以了，即運(yùn)行時(shí)計(jì)算機(jī)的處理器數(shù)量。也可以通過(guò)設(shè)置系統(tǒng)屬性：-Djava.util.concurrent .ForkJoinPool.common.parallelism=N?（N為線(xiàn)程數(shù)量）,來(lái)調(diào)整ForkJoinPool的線(xiàn)程數(shù)量。

值得注意的是，當(dāng)前執(zhí)行的線(xiàn)程也會(huì)被用來(lái)執(zhí)行任務(wù)，所以最終的線(xiàn)程個(gè)數(shù)為N+1，1就是當(dāng)前的主線(xiàn)程。

這里就有一個(gè)問(wèn)題，如果你在并行流的執(zhí)行計(jì)算使用了_阻塞操作_，如I/O，那么很可能會(huì)導(dǎo)致一些問(wèn)題：

public?static?String?query(String?question)?{
??List?engines?=?new?ArrayList();
??engines.add("http://www.google.com/?q=");
??engines.add("http://duckduckgo.com/?q=");
??engines.add("http://www.bing.com/search?q=");
???
??//?get?element?as?soon?as?it?is?available
??Optional?result?=?engines.stream().parallel().map((base)?-?{
????String?url?=?base?+?question;
????//?open?connection?and?fetch?the?result
????return?WS.url(url).get();
??}).findAny();
??return?result.get();
}

這個(gè)例子很典型，讓我們來(lái)分析一下：

• 這個(gè)并行流計(jì)算操作將由主線(xiàn)程和JVM默認(rèn)的ForkJoinPool.commonPool()來(lái)共同執(zhí)行。

• map中是一個(gè)阻塞方法，需要通過(guò)訪(fǎng)問(wèn)HTTP接口并得到它的response，所以任何一個(gè)worker線(xiàn)程在執(zhí)行到這里的時(shí)候都會(huì)阻塞并等待結(jié)果。

• 所以當(dāng)此時(shí)再其他地方通過(guò)并行流方式調(diào)用計(jì)算方法的時(shí)候，將會(huì)受到此處阻塞等待的方法的影響。

• 目前的ForkJoinPool的實(shí)現(xiàn)并未考慮補(bǔ)償?shù)却切┳枞诘却律傻木€(xiàn)程的工作worker線(xiàn)程，所以最終ForkJoinPool.commonPool()中的線(xiàn)程將備用光并且阻塞等待。

正如我們上面那個(gè)列子的情況分析得知，lambda的執(zhí)行并不是瞬間完成的,所有使用parallel streams的程序都有可能成為阻塞程序的源頭， 并且在執(zhí)行過(guò)程中程序中的其他部分將無(wú)法訪(fǎng)問(wèn)這些workers，這意味著任何依賴(lài)parallel streams的程序在什么別的東西占用著common ForkJoinPool時(shí)將會(huì)變得不可預(yù)知并且暗藏危機(jī)。

小結(jié)：

1. 當(dāng)需要處理遞歸分治算法時(shí)，考慮使用ForkJoinPool。

2. 仔細(xì)設(shè)置不再進(jìn)行任務(wù)劃分的閾值，這個(gè)閾值對(duì)性能有影響。

3. Java 8中的一些特性會(huì)使用到ForkJoinPool中的通用線(xiàn)程池。在某些場(chǎng)合下，需要調(diào)整該線(xiàn)程池的默認(rèn)的線(xiàn)程數(shù)量

4. lambda應(yīng)該盡量避免副作用，也就是說(shuō)，避免突變基于堆的狀態(tài)以及任何IO

5. lambda應(yīng)該互不干擾，也就是說(shuō)避免修改數(shù)據(jù)源（因?yàn)檫@可能帶來(lái)線(xiàn)程安全的問(wèn)題）

6. 避免訪(fǎng)問(wèn)在流操作生命周期內(nèi)可能會(huì)改變的狀態(tài)

并行流的性能

并行流框架的性能受以下因素影響：

• 數(shù)據(jù)大小：數(shù)據(jù)夠大，每個(gè)管道處理時(shí)間夠長(zhǎng)，并行才有意義；

• 源數(shù)據(jù)結(jié)構(gòu)：每個(gè)管道操作都是基于初始數(shù)據(jù)源，通常是集合，將不同的集合數(shù)據(jù)源分割會(huì)有一定消耗；

• 裝箱：處理基本類(lèi)型比裝箱類(lèi)型要快；

• 核的數(shù)量：默認(rèn)情況下，核數(shù)量越多，底層fork/join線(xiàn)程池啟動(dòng)線(xiàn)程就越多；

• 單元處理開(kāi)銷(xiāo)：花在流中每個(gè)元素身上的時(shí)間越長(zhǎng)，并行操作帶來(lái)的性能提升越明顯；

源數(shù)據(jù)結(jié)構(gòu)分為以下3組：

• 性能好：ArrayList、數(shù)組或IntStream.range(數(shù)據(jù)支持隨機(jī)讀取，能輕易地被任意分割)

• 性能一般：HashSet、TreeSet(數(shù)據(jù)不易公平地分解，大部分也是可以的)

• 性能差：LinkedList(需要遍歷鏈表，難以對(duì)半分解)、Stream.iterate和BufferedReader.lines(長(zhǎng)度未知，難以分解)

注意：下面幾個(gè)部分節(jié)選自：Streams 的幕后原理，順便感謝一下作者_(dá)Brian Goetz_，寫(xiě)的太通透了。

NQ模型

要確定并行性是否會(huì)帶來(lái)提速，需要考慮的最后兩個(gè)因素是：可用的數(shù)據(jù)量和針對(duì)每個(gè)數(shù)據(jù)元素執(zhí)行的計(jì)算量。

在我們最初的并行分解描述中，我們采用的概念是拆分來(lái)源，直到分段足夠小，以致解決該分段上的問(wèn)題的順序方法更高效。分段大小必須依賴(lài)于所解決的問(wèn)題，確切的講，取決于每個(gè)元素完成的工作量。例如，計(jì)算一個(gè)字符串的長(zhǎng)度涉及的工作比計(jì)算字符串的?SHA-1?哈希值要少得多。為每個(gè)元素完成的工作越多，“大到足夠利用并行性” 的閾值就越低。類(lèi)似地，擁有的數(shù)據(jù)越多， 拆分的分段就越多，而不會(huì)與 “太小” 閾值發(fā)生沖突。

一個(gè)簡(jiǎn)單但有用的并行性能模型是?NQ?模型，其中?N?是數(shù)據(jù)元素?cái)?shù)量，Q?是為每個(gè)元素執(zhí)行的工作量。乘積?N*Q?越大，就越有可能獲得并行提速。對(duì)于具有很小的?Q?的問(wèn)題，比如對(duì)數(shù)字求和，您通?？赡芟Ｍ吹?N > 10,000?以獲得提速；隨著?Q?增加，獲得提速所需的數(shù)據(jù)大小將會(huì)減小。

并行化的許多阻礙（比如拆分成本、組合成本或遇到順序敏感性）都可以通過(guò)?Q?更高的操作來(lái)緩解。盡管拆分某個(gè)?LinkedList?特征的結(jié)果可能很糟糕，但只要擁有足夠大的?Q，仍然可能獲得并行提速。

遇到順序

遇到順序指的是來(lái)源分發(fā)元素的順序是否對(duì)計(jì)算至關(guān)重要。一些來(lái)源（比如基于哈希的集合和映射）沒(méi)有有意義的遇到順序。流標(biāo)志?ORDERED?描述了流是否有有意義的遇到順序。JDK 集合的?spliterator?會(huì)根據(jù)集合的規(guī)范來(lái)設(shè)置此標(biāo)志；一些中間操作可能注入?ORDERED?(sorted()) 或清除它 (unordered())。

如果流沒(méi)有遇到順序，大部分流操作都必須遵守該順序。對(duì)于順序執(zhí)行，會(huì)「自動(dòng)保留遇到順序」，因?yàn)樵貢?huì)按遇到它們的順序自然地處理。甚至在并行執(zhí)行中，許多操作（無(wú)狀態(tài)中間操作和一些終止操作（比如?reduce()）），遵守遇到順序不會(huì)產(chǎn)生任何實(shí)際成本。但對(duì)于其他操作（有狀態(tài)中間操作，其語(yǔ)義與遇到順序關(guān)聯(lián)的終止操作，比如?findFirst()?或?forEachOrdered()）， 在并行執(zhí)行中遵守遇到順序的責(zé)任可能很重大。如果流有一個(gè)已定義的遇到順序，但該順序?qū)Y(jié)果沒(méi)有意義， 那么可以通過(guò)使用?unordered()?操作刪除?ORDERED?標(biāo)志，加速包含順序敏感型操作的管道的順序執(zhí)行。

作為對(duì)遇到順序敏感的操作的示例，可以考慮?limit()，它會(huì)在指定大小處截?cái)嘁粋€(gè)流。在順序執(zhí)行中實(shí)現(xiàn)?limit()?很簡(jiǎn)單：保留一個(gè)已看到多少元素的計(jì)數(shù)器，在這之后丟棄任何元素。但是在并行執(zhí)行中，實(shí)現(xiàn)?limit()?要復(fù)雜得多；您需要保留前?N?個(gè)元素。此要求大大限制了利用并行性的能力；如果輸入劃分為多個(gè)部分，您只有在某個(gè)部分之前的所有部分都已完成后，才知道該部分的結(jié)果是否將包含在最終結(jié)果中。因此，該實(shí)現(xiàn)一般會(huì)錯(cuò)誤地選擇不使用所有可用的核心，或者緩存整個(gè)試驗(yàn)性結(jié)果，直到您達(dá)到目標(biāo)長(zhǎng)度。

如果流沒(méi)有遇到順序，limit()?操作可以自由選擇任何?N?個(gè)元素，這讓執(zhí)行效率變得高得多。知道元素后可立即將其發(fā)往下游， 無(wú)需任何緩存，而且線(xiàn)程之間唯一需要執(zhí)行的協(xié)調(diào)是發(fā)送一個(gè)信號(hào)來(lái)確保未超出目標(biāo)流長(zhǎng)度。

遇到順序成本的另一個(gè)不太常見(jiàn)的示例是排序。如果遇到順序有意義，那么?sorted()?操作會(huì)實(shí)現(xiàn)一種穩(wěn)定 排序 （相同的元素按照它們進(jìn)入輸入時(shí)的相同順序出現(xiàn)在輸出中），而對(duì)于無(wú)序的流，穩(wěn)定性（具有成本）不是必需的。?distinct()?具有類(lèi)似的情況：如果流有一個(gè)遇到順序，那么對(duì)于多個(gè)相同的輸入元素，distinct()?必須發(fā)出其中的第一個(gè)， 而對(duì)于無(wú)序的流，它可以發(fā)出任何元素 — 同樣可以獲得高效得多的并行實(shí)現(xiàn)。

在您使用?collect()?聚合時(shí)會(huì)遇到類(lèi)似的情形。如果在無(wú)序流上執(zhí)行?collect(groupingBy()) 操作， 與任何鍵對(duì)應(yīng)的元素都必須按它們?cè)谳斎胫谐霈F(xiàn)的順序提供給下游收集器。此順序?qū)?yīng)用程序通常沒(méi)有什么意義，而且任何順序都沒(méi)有意義。在這些情況下，可能最好選擇一個(gè)并發(fā) 收集器（比如?groupingByConcurrent()），它可以忽略遇到順序， 并讓所有線(xiàn)程直接收集到一個(gè)共享的并發(fā)數(shù)據(jù)結(jié)構(gòu)中（比如?ConcurrentHashMap），而不是讓每個(gè)線(xiàn)程收集到它自己的中間映射中， 然后再合并中間映射（這可能產(chǎn)生很高的成本）。

什么時(shí)候該使用并行流

談了這么多，關(guān)于并行流parallelStream的使用注意事項(xiàng)需要格外注意，它并不是解決性能的萬(wàn)金油，相反，如果使用不當(dāng)會(huì)嚴(yán)重影響性能。我會(huì)在另外一篇文章里單獨(dú)談這個(gè)問(wèn)題。

References

• http://movingon.cn/2017/05/02/jdk8-Stream-BaseStream-%E6%BA%90%E7%A0%81%E9%9A%BE%E7%82%B9%E6%B5%85%E6%9E%901/

• https://www.jianshu.com/p/bd825cb89e00

• https://jrebel.com/rebellabs/java-parallel-streams-are-bad-for-your-health/

• https://blog.csdn.net/weixx3/article/details/81266552

• https://www.ibm.com/developerworks/cn/java/j-java-streams-5-brian-goetz/index.html

• https://www.ibm.com/developerworks/cn/java/j-java-streams-3-brian-goetz/index.html

• https://juejin.im/post/5dc5a148f265da4d4f65c191

• https://stackoverrun.com/cn/q/10341100