Java8 中的 Stream 那么強大，那你知道它的原理是什么嗎？

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

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

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

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

Stream的組成與特點

• 元素是特定類型的對象，形成一個隊列。?Java中的Stream并_不會_向集合那樣存儲和管理元素，而是按需計算

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

• 聚合操作類似SQL語句一樣的操作， 比如filter,?map,?reduce,?find,?match,?sorted等

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

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

1.0-1.4 中的 java.lang.Thread

5.0 中的 java.util.concurrent

6.0 中的 Phasers 等

7.0 中的 Fork/Join 框架

8.0 中的 Lambda

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

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

BaseStream接口

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();
}

S extends BaseStream

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

Stream接口

public?interface?Stream?extends?BaseStream>?

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

關(guān)閉流操作

/**?*?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);

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

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

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

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

并行流和串行流

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.

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

ParallelStream背后的男人：ForkJoinPool

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

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. 當只剩下最后一個任務(wù)時，還是會存在競爭，是通過CAS來實現(xiàn)的；

用ForkJoinPool的眼光來看ParallelStream

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

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();
}

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

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

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

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

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

1. 當需要處理遞歸分治算法時，考慮使用ForkJoinPool。

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

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

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

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

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

并行流的性能

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

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

• 裝箱：處理基本類型比裝箱類型要快；

• 核的數(shù)量：默認情況下，核數(shù)量越多，底層fork/join線程池啟動線程就越多；

• 單元處理開銷：花在流中每個元素身上的時間越長，并行操作帶來的性能提升越明顯；

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

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

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

NQ模型

遇到順序

什么時候該使用并行流

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

面試官：Java 反射是什么？
Java 17：和遺留 25 年的漏洞 Say Goodbye
Java 框架 Mybatis 插件開發(fā)指南，超詳細！
18 個 Java8 處理日期的新花樣，肯定沒用過！
你還在 new 對象嗎？Java8 通用 Builder 了解一下