[[382101]]

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

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

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

本文會對Stream的實現(xiàn)原理進行剖析。

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操作還有兩個基礎(chǔ)的特征：

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

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

Stream 的并行操作依賴于 Java7 中引入的 Fork/Join 框架（JSR166y）來拆分任務(wù)和加速處理過程。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具有平行處理能力，處理的過程會分而治之，也就是將一個大任務(wù)切分成多個小任務(wù)，這表示每個任務(wù)都是一個操作： 

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

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

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

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

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

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

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實例，也就是說它們分別支持對當(dāng)前流進行串行或者并行的操作，并返回「改變」后的流對象。

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

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

Stream接口

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

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

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

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);  
    ... 

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

關(guān)閉流操作

BaseStream 實現(xiàn)了 AutoCloseable 接口，也就是 close() 方法會在流關(guān)閉時被調(diào)用。同時，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. * * <p>This is an <a href="package-summary.html#StreamOps">intermediate * operation</a>. * * @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)用的時候會觸發(fā)調(diào)用流對象的onClose()方法，但有幾點需要注意：

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

并行流和串行流

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

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

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

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

ParallelStream背后的男人：ForkJoinPool

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

ForkJoinPool主要用來使用分治法(Divide-and-Conquer Algorithm) 來解決問題，典型的應(yīng)用比如快速排序算法。這里的要點在于，F(xiàn)orkJoinPool需要使用相對少的線程來處理大量的任務(wù)。比如要對1000萬個數(shù)據(jù)進行排序，那么會將這個任務(wù)分割成兩個500 萬的排序任務(wù)和一個針對這兩組500萬數(shù)據(jù)的合并任務(wù)。以此類推，對于500萬的數(shù)據(jù)也會做出同樣的分割處理，到最后會設(shè)置一個閾值來規(guī)定當(dāng)數(shù)據(jù)規(guī)模到多少時，停止這樣的分割處理。比如，當(dāng)元素的數(shù)量小于10時，會停止分割，轉(zhuǎn)而使用插入排序?qū)λ鼈冞M行排序。那么到最后，所有的任務(wù)加起來會有大概2000000+個。

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

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

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

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

Work Stealing原理：

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

用ForkJoinPool的眼光來看ParallelStream

Java 8為ForkJoinPool添加了一個通用線程池，這個線程池用來處理那些沒有被顯式提交到任何線程池的任務(wù)。它是ForkJoinPool類型上的一個靜態(tài)元素，它擁有的默認線程數(shù)量等于運行計算機上的CPU數(shù)量。當(dāng)調(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)建一個任務(wù)，該任務(wù)會被前文中提到的ForkJoinPool中的commonPool處理。以上的并行計算邏輯當(dāng)然也可以使用ThreadPoolExecutor完成，但是就代碼的可讀性和代碼量而言，使用ForkJoinPool明顯更勝一籌。

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

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

這里就有一個問題，如果你在并行流的執(zhí)行計算使用了阻塞操作，如I/O，那么很可能會導(dǎ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í)行到這里的時候都會阻塞并等待結(jié)果。
•  所以當(dāng)此時再其他地方通過并行流方式調(diào)用計算方法的時候，將會受到此處阻塞等待的方法的影響。
•  目前的ForkJoinPool的實現(xiàn)并未考慮補償?shù)却切┳枞诘却律傻木€程的工作worker線程，所以最終ForkJoinPool.commonPool()中的線程將備用光并且阻塞等待。

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

小結(jié)：

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

仔細設(shè)置不再進行任務(wù)劃分的閾值，這個閾值對性能有影響。

 Java 8中的一些特性會使用到ForkJoinPool中的通用線程池。在某些場合下，需要調(diào)整該線程池的默認的線程數(shù)量

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

 lambda應(yīng)該互不干擾，也就是說避免修改數(shù)據(jù)源（因為這可能帶來線程安全的問題）

 避免訪問在流操作生命周期內(nèi)可能會改變的狀態(tài)

并行流的性能

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

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

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

•  性能好：ArrayList、數(shù)組或IntStream.range(數(shù)據(jù)支持隨機讀取，能輕易地被任意分割)
•  性能一般：HashSet、TreeSet(數(shù)據(jù)不易公平地分解，大部分也是可以的)
•  性能差：LinkedList(需要遍歷鏈表，難以對半分解)、Stream.iterate和BufferedReader.lines(長度未知，難以分解)

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

NQ模型

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

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

一個簡單但有用的并行性能模型是 NQ 模型，其中 N 是數(shù)據(jù)元素數(shù)量，Q 是為每個元素執(zhí)行的工作量。乘積 N*Q 越大，就越有可能獲得并行提速。對于具有很小的 Q 的問題，比如對數(shù)字求和，您通常可能希望看到 N > 10,000 以獲得提速；隨著 Q 增加，獲得提速所需的數(shù)據(jù)大小將會減小。

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

遇到順序

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

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

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

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

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

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

什么時候該使用并行流

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