前言:

Stream 使用一種類似用 SQL 語句從數(shù)據(jù)庫查詢數(shù)據(jù)的直觀方式來提供一種對 Java 集合運算和表達的高階抽象。

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

1、Stream的組成與特點

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

• 元素是特定類型的對象，形成一個隊列。Java中的Stream并_不會_向集合那樣存儲和管理元素，而是按需計算
• 數(shù)據(jù)源流的來源可以是集合Collection、數(shù)組Array、I/O channel， 產(chǎn)生器generator 等
• 聚合操作類似SQL語句一樣的操作， 比如filter, map, reduce, find, match, sorted等

和以前的Collection操作不同， Stream操作還有兩個基礎的特征：

• Pipelining: 中間操作都會返回流對象本身。這樣多個操作可以串聯(lián)成一個管道， 如同流式風格（fluent style）。這樣做可以對操作進行優(yōu)化， 比如延遲執(zhí)行(laziness evaluation)和短路( short-circuiting)
• 內(nèi)部迭代：以前對集合遍歷都是通過Iterator或者For-Each的方式, 顯式的在集合外部進行迭代， 這叫做外部迭代。Stream提供了內(nèi)部迭代的方式， 通過訪問者模式 (Visitor)實現(xiàn)。

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

Stream 的并行操作依賴于 Java7 中引入的 Fork/Join 框架（JSR166y）來拆分任務和加速處理過程。

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具有平行處理能力，處理的過程會分而治之，也就是將一個大任務切分成多個小任務，這表示每個任務都是一個操作：

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

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

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

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

這里有一個疑問，如果結果需要有序，是否和我們的并行執(zhí)行的初衷相悖？是的，這個場景下明顯無需使用并行流，直接用串行流執(zhí)行即可， 否則性能可能更差，因為最后又強行將所有并行結果進行了排序。

OK，下面我們先介紹一下Stream接口的相關知識。

2、BaseStream接口

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

public interface BaseStream<T, S extends BaseStream<T, S>>        extends AutoCloseable {    Iterator<T> iterator();    Spliterator<T> spliterator();    boolean isParallel();    S sequential();    S parallel();    S unordered();    S onClose(Runnable closeHandler);    void close();}

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

S extends BaseStream<T, S>

是不是有點暈？

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

“

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

”

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

3、Stream接口

再來看一下Stream的接口聲明：

public interface Stream<T> extends BaseStream<T, Stream<T>>

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

Stream<T> filter(Predicate<? super T> predicate);<R> Stream<R> map(Function<? super T, ? extends R> mapper);<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);Stream<T> sorted();Stream<T> peek(Consumer<? super T> action);Stream<T> limit(long maxSize);Stream<T> skip(long n);...

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

4、關閉流操作

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

S onClose(Runnable closeHandler);

當AutoCloseable的close()接口被調(diào)用的時候會觸發(fā)調(diào)用流對象的onClose()方法，

但有幾點需要注意：

• onClose() 方法會返回流對象本身，也就是說可以對改對象進行多次調(diào)用
• 如果調(diào)用了多個onClose() 方法，它會按照調(diào)用的順序觸發(fā)，但是如果某個方法有異常則只會向上拋出第一個異常
• 前一個 onClose() 方法拋出了異常不會影響后續(xù) onClose() 方法的使用
• 如果多個 onClose() 方法都拋出異常，只展示第一個異常的堆棧，而其他異常會被壓縮，只展示部分信息

5、并行流和串行流

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

參考parallel()方法的說明：

“

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)用同樣的方法并不會生成新的流，而是直接復用當前的流對象。

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

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

6、ParallelStream背后的男人：ForkJoinPool

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

ForkJoinPool主要用來使用分治法(Divide-and-Conquer Algorithm) 來解決問題，典型的應用比如_快速排序算法_。這里的要點在于，ForkJoinPool需要使用相對少的線程來處理大量的任務。

比如要對1000萬個數(shù)據(jù)進行排序，那么會將這個任務分割成兩個500 萬的排序任務和一個針對這兩組500萬數(shù)據(jù)的合并任務。

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

“

問題的關鍵在于，對于一個任務而言，只有當它所有的子任務完成之后，它才能夠被執(zhí)行，想象一下歸并排序的過程。

”

所以當使用ThreadPoolExecutor時，使用分治法會存在問題，因為ThreadPoolExecutor中的線程無法向 任務隊列中再添加一個任務并且在等待該任務完成之后再繼續(xù)執(zhí)行。而使用ForkJoinPool時，就能夠讓其中的線程創(chuàng)建新的任務，并掛起當前的任務，此時線程就能夠從隊列中選擇子任務執(zhí)行。

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

首先，使用ForkJoinPool能夠使用數(shù)量有限的線程來完成非常多的具有「父子關系」的任務，比如使用4個線程來完成超過200萬個任務。使用ThreadPoolExecutor 時，是不可能完成的，因為ThreadPoolExecutor中的Thread無法選擇優(yōu)先執(zhí)行子任務，需要完成200萬個具有父子關系的任務時，也需要200萬個線程，顯然這是不可行的。

Work Stealing原理：

• 每個工作線程都有自己的工作隊列WorkQueue；
• 這是一個雙端隊列dequeue，它是線程私有的；
• ForkJoinTask中fork的子任務，將放入運行該任務的工作線程的隊頭，工作線程將以LIFO的順序來處理工作隊列中的任務，即堆棧的方式；
• 為了最大化地利用CPU，空閑的線程將從其它線程的隊列中「竊取」任務來執(zhí)行
• 但是是從工作隊列的尾部竊取任務，以減少和隊列所屬線程之間的競爭；
• 雙端隊列的操作：push()/pop()僅在其所有者工作線程中調(diào)用，poll()是由其它線程竊取任務時調(diào)用的；
• 當只剩下最后一個任務時，還是會存在競爭，是通過CAS來實現(xiàn)的；

7、用ForkJoinPool的眼光來看ParallelStream

Java 8為ForkJoinPool添加了一個通用線程池，這個線程池用來處理那些沒有被顯式提交到任何線程池的任務。它是ForkJoinPool類型上的一個靜態(tài)元素，它擁有的默認線程數(shù)量等于運行計算機上的CPU數(shù)量。

當調(diào)用Arrays 類上添加的新方法時，自動并行化就會發(fā)生。

比如用來排序一個數(shù)組的并行快速排序，用來對一個數(shù)組中的元素進行并行遍歷。自動并行化也被運用在Java 8新添加的Stream API中。

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

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

對于列表中的元素的操作都會以并行的方式執(zhí)行。forEach方法會為每個元素的計算操作創(chuàng)建一個任務，該任務會被前文中提到的ForkJoinPool中的commonPool處理。

以上的并行計算邏輯當然也可以使用ThreadPoolExecutor完成，但是就代碼的可讀性和代碼量而言，使用ForkJoinPool明顯更勝一籌。

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

值得注意的是:當前執(zhí)行的線程也會被用來執(zhí)行任務，所以最終的線程個數(shù)為N+1，1就是當前的主線程。

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

public static String query(String question) {  List<String> engines = new ArrayList<String>();  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<String> 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();}

這個例子很典型，讓我們來分析一下：

這個并行流計算操作將由主線程和JVM默認的ForkJoinPool.commonPool()來共同執(zhí)行。

map中是一個阻塞方法，需要通過訪問HTTP接口并得到它的response，所以任何一個worker線程在執(zhí)行到這里的時候都會阻塞并等待結果。

所以當此時再其他地方通過并行流方式調(diào)用計算方法的時候，將會受到此處阻塞等待的方法的影響。

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

?“

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

”

小結：

• 當需要處理遞歸分治算法時，考慮使用ForkJoinPool。
• 仔細設置不再進行任務劃分的閾值，這個閾值對性能有影響。
• Java 8中的一些特性會使用到ForkJoinPool中的通用線程池。在某些場合下，需要調(diào)整該線程池的默認的線程數(shù)量
• lambda應該盡量避免副作用，也就是說，避免突變基于堆的狀態(tài)以及任何IO
• lambda應該互不干擾，也就是說避免修改數(shù)據(jù)源（因為這可能帶來線程安全的問題）
• 避免訪問在流操作生命周期內(nèi)可能會改變的狀態(tài)

8、并行流的性能

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

• 數(shù)據(jù)大?。?/strong>數(shù)據(jù)夠大，每個管道處理時間夠長，并行才有意義；
• 源數(shù)據(jù)結構：每個管道操作都是基于初始數(shù)據(jù)源，通常是集合，將不同的集合數(shù)據(jù)源分割會有一定消耗；
• 裝箱：處理基本類型比裝箱類型要快；
• 核的數(shù)量：默認情況下，核數(shù)量越多，底層fork/join線程池啟動線程就越多；
• 單元處理開銷：花在流中每個元素身上的時間越長，并行操作帶來的性能提升越明顯；