原文：https://blog.csdn.net/u011001723/article/details/52794455/
一、測試一個案例，說明java8的流操作是並行操作
1、代碼
package com.spring.test.service.forkjoin; import java.util.ArrayList; import java.util.List; /** * */ public class Java8Test { public static void main(String[] args) { testList(); } public static void testList(){ //源數據 List<Integer> integerList=new ArrayList<>(); integerList.add(0); integerList.add(1); integerList.add(2); integerList.add(3); integerList.add(4); integerList.add(5); integerList.add(6); integerList.add(7); integerList.add(8); integerList.add(9); //java的順序操作 for(Integer a:integerList){ System.out.println(Thread.currentThread().getName()+"==>"+a); } //java的並行操作 integerList.parallelStream().forEach(integer -> { System.out.println(Thread.currentThread().getName()+"==>"+integer); }); } }
2、執行結果
main==>0 main==>1 main==>2 main==>3 main==>4 main==>5 main==>6 main==>7 main==>8 main==>9 main==>6 main==>5 ForkJoinPool.commonPool-worker-4==>7 ForkJoinPool.commonPool-worker-1==>2 ForkJoinPool.commonPool-worker-4==>9 ForkJoinPool.commonPool-worker-2==>8 ForkJoinPool.commonPool-worker-4==>0 ForkJoinPool.commonPool-worker-3==>1 ForkJoinPool.commonPool-worker-5==>4 ForkJoinPool.commonPool-worker-1==>3
二、parallelStream背后的男人:ForkJoinPool
要想深入的研究parallelStream之前,那么我們必須先了解ForkJoin框架和ForkJoinPool.本文旨在parallelStream,但因為兩種關系甚密,故在此簡單介紹一下ForkJoinPool,如有興趣可以更深入的去了解下ForkJoin***(當然,如果你想真正的搞透parallelStream,那么你依然需要先搞透ForkJoinPool).*
ForkJoin框架是從jdk7中新特性,它同ThreadPoolExecutor一樣，也實現了Executor和ExecutorService接口。它使用了一個無限隊列來保存需要執行的任務，而線程的數量則是通過構造函數傳入，如果沒有向構造函數中傳入希望的線程數量，那么當前計算機可用的CPU數量會被設置為線程數量作為默認值。
ForkJoinPool主要用來使用分治法(Divide-and-Conquer Algorithm)來解決問題。典型的應用比如快速排序算法。這里的要點在於，ForkJoinPool需要使用相對少的線程來處理大量的任務。比如要對1000萬個數據進行排序，那么會將這個任務分割成兩個500萬的排序任務和一個針對這兩組500萬數據的合並任務。以此類推，對於500萬的數據也會做出同樣的分割處理，到最后會設置一個閾值來規定當數據規模到多少時，停止這樣的分割處理。比如，當元素的數量小於10時，會停止分割，轉而使用插入排序對它們進行排序。那么到最后，所有的任務加起來會有大概2000000+個。問題的關鍵在於，對於一個任務而言，只有當它所有的子任務完成之后，它才能夠被執行。
所以當使用ThreadPoolExecutor時，使用分治法會存在問題，因為ThreadPoolExecutor中的線程無法像任務隊列中再添加一個任務並且在等待該任務完成之后再繼續執行。而使用ForkJoinPool時，就能夠讓其中的線程創建新的任務，並掛起當前的任務，此時線程就能夠從隊列中選擇子任務執行。
那么使用ThreadPoolExecutor或者ForkJoinPool，會有什么性能的差異呢？
首先，使用ForkJoinPool能夠使用數量有限的線程來完成非常多的具有父子關系的任務，比如使用4個線程來完成超過200萬個任務。但是，使用ThreadPoolExecutor時，是不可能完成的，因為ThreadPoolExecutor中的Thread無法選擇優先執行子任務，需要完成200萬個具有父子關系的任務時，也需要200萬個線程，顯然這是不可行的。
工作竊取算法
forkjoin最核心的地方就是利用了現代硬件設備多核,在一個操作時候會有空閑的cpu,那么如何利用好這個空閑的cpu就成了提高性能的關鍵,而這里我們要提到的工作竊取（work-stealing）算法就是整個forkjion框架的核心理念,工作竊取（work-stealing）算法是指某個線程從其他隊列里竊取任務來執行。
那么為什么需要使用工作竊取算法呢？
假如我們需要做一個比較大的任務，我們可以把這個任務分割為若干互不依賴的子任務，為了減少線程間的競爭，於是把這些子任務分別放到不同的隊列里，並為每個隊列創建一個單獨的線程來執行隊列里的任務，線程和隊列一一對應，比如A線程負責處理A隊列里的任務。但是有的線程會先把自己隊列里的任務干完，而其他線程對應的隊列里還有任務等待處理。干完活的線程與其等着，不如去幫其他線程干活，於是它就去其他線程的隊列里竊取一個任務來執行。而在這時它們會訪問同一個隊列，所以為了減少竊取任務線程和被竊取任務線程之間的競爭，通常會使用雙端隊列，被竊取任務線程永遠從雙端隊列的頭部拿任務執行，而竊取任務的線程永遠從雙端隊列的尾部拿任務執行。
工作竊取算法的優點是充分利用線程進行並行計算，並減少了線程間的競爭，其缺點是在某些情況下還是存在競爭，比如雙端隊列里只有一個任務時。並且消耗了更多的系統資源，比如創建多個線程和多個雙端隊列。
用看forkjion的眼光來看ParallelStreams
上文中已經提到了在Java 8引入了自動並行化的概念。它能夠讓一部分Java代碼自動地以並行的方式執行，也就是我們使用了ForkJoinPool的ParallelStream。
Java 8為ForkJoinPool添加了一個通用線程池，這個線程池用來處理那些沒有被顯式提交到任何線程池的任務。它是ForkJoinPool類型上的一個靜態元素，它擁有的默認線程數量等於運行計算機上的處理器數量。當調用Arrays類上添加的新方法時，自動並行化就會發生。比如用來排序一個數組的並行快速排序，用來對一個數組中的元素進行並行遍歷。自動並行化也被運用在Java 8新添加的Stream API中。
比如下面的代碼用來遍歷列表中的元素並執行需要的操作：
List<UserInfo> userInfoList = DaoContainers.getUserInfoDAO().queryAllByList(new UserInfoModel()); userInfoList.parallelStream().forEach(RedisUserApi::setUserIdUserInfo);
對於列表中的元素的操作都會以並行的方式執行。forEach方法會為每個元素的計算操作創建一個任務，該任務會被前文中提到的ForkJoinPool中的通用線程池處理。以上的並行計算邏輯當然也可以使用ThreadPoolExecutor完成，但是就代碼的可讀性和代碼量而言，使用ForkJoinPool明顯更勝一籌。
對於ForkJoinPool通用線程池的線程數量，通常使用默認值就可以了，即運行時計算機的處理器數量。我這里提供了一個示例的代碼讓你了解jvm所使用的ForkJoinPool的線程數量, 你可以可以通過設置系統屬性：-Djava.util.concurrent.ForkJoinPool.common.parallelism=N （N為線程數量）,來調整ForkJoinPool的線程數量,可以嘗試調整成不同的參數來觀察每次的輸出結果:
import java.util.ArrayList; import java.util.List; import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.CountDownLatch; /** * @description 這是一個用來讓你更加熟悉parallelStream的原理的實力 * @date 2016年10月11日18:26:55 * @version v1.0 * @author wangguangdong */ public class App { public static void main(String[] args) throws Exception { System.out.println("Hello World!"); // 構造一個10000個元素的集合 List<Integer> list = new ArrayList<>(); for (int i = 0; i < 10000; i++) { list.add(i); } // 統計並行執行list的線程 Set<Thread> threadSet = new CopyOnWriteArraySet<>(); // 並行執行 list.parallelStream().forEach(integer -> { Thread thread = Thread.currentThread(); // System.out.println(thread); // 統計並行執行list的線程 threadSet.add(thread); }); System.out.println("threadSet一共有" + threadSet.size() + "個線程"); System.out.println("系統一個有"+Runtime.getRuntime().availableProcessors()+"個cpu"); List<Integer> list1 = new ArrayList<>(); List<Integer> list2 = new ArrayList<>(); for (int i = 0; i < 100000; i++) { list1.add(i); list2.add(i); } Set<Thread> threadSetTwo = new CopyOnWriteArraySet<>(); CountDownLatch countDownLatch = new CountDownLatch(2); Thread threadA = new Thread(() -> { list1.parallelStream().forEach(integer -> { Thread thread = Thread.currentThread(); // System.out.println("list1" + thread); threadSetTwo.add(thread); }); countDownLatch.countDown(); }); Thread threadB = new Thread(() -> { list2.parallelStream().forEach(integer -> { Thread thread = Thread.currentThread(); // System.out.println("list2" + thread); threadSetTwo.add(thread); }); countDownLatch.countDown(); }); threadA.start(); threadB.start(); countDownLatch.await(); System.out.print("threadSetTwo一共有" + threadSetTwo.size() + "個線程"); System.out.println("---------------------------"); System.out.println(threadSet); System.out.println(threadSetTwo); System.out.println("---------------------------"); threadSetTwo.addAll(threadSet); System.out.println(threadSetTwo); System.out.println("threadSetTwo一共有" + threadSetTwo.size() + "個線程"); System.out.println("系統一個有"+Runtime.getRuntime().availableProcessors()+"個cpu"); } }
出現這種現象的原因是，forEach方法用了一些小把戲。它會將執行forEach本身的線程也作為線程池中的一個工作線程。因此，即使將ForkJoinPool的通用線程池的線程數量設置為1，實際上也會有2個工作線程。因此在使用forEach的時候，線程數為1的ForkJoinPool通用線程池和線程數為2的ThreadPoolExecutor是等價的。
所以當ForkJoinPool通用線程池實際需要4個工作線程時，可以將它設置成3，那么在運行時可用的工作線程就是4了。
小結:
1. 當需要處理遞歸分治算法時，考慮使用ForkJoinPool。
2. 仔細設置不再進行任務划分的閾值，這個閾值對性能有影響。
3. Java 8中的一些特性會使用到ForkJoinPool中的通用線程池。在某些場合下，需要調整該線程池的默認的線程數量。
ParallelStreams 的陷阱
上文中我們已經看到了ParallelStream他強大無比的特性,但這里我們就講告訴你ParallelStreams不是萬金油,而是一把雙刃劍,如果錯誤的使用反倒可能傷人傷己.
以下是一個我們項目里使用 parallel streams 的很常見的情況。在這個例子中，我們想同時調用不同地址的api中並且獲得第一個返回的結果。
public static String query(String q, List<String> engines) { Optional<String> result = engines.stream().parallel().map((base) -> { String url = base + q; return WS.url(url).get(); }).findAny(); return result.get(); }
可能有很多朋友在jdk7用future配合countDownLatch自己實現的這個功能,但是jdk8的朋友基本都會用上面的實現方式,那么自信深究一下究竟自己用future實現的這個功能和利用jdk8的parallelStream來實現這個功能有什么不同點呢?坑又在哪里呢？
讓我們細思思考一下整個功能究竟是如何運轉的。首先我們的集合元素engines 由ParallelStreams並行的去進行map操作(ParallelStreams使用JVM默認的forkJoin框架的線程池由當前線程去執行並行操作).
然而,這里需要注意的一地方是我們在調用第三方的api請求是一個響應略慢而且會阻塞操作的一個過程。所以在某時刻所有線程都會調用 get() 方法並且在那里等待結果返回.
再回過頭仔細思考一下這個功能的實現過程是我們一開始想要的嗎？我們是在同一時間等待所有的結果,而不是遍歷這個列表按順序等待每個回答.然而，由於ForkJoinPool workders的存在，這樣平行的等待相對於使用主線程的等待會產生的一種副作用.
現在ForkJoin pool (關於forkjion的更多實現你可以去搜索引擎中去看一下他的具體實現方式) 的實現是: 它並不會因為產生了新的workers而抵消掉阻塞的workers。那么在某個時間所有 ForkJoinPool.common() 的線程都會被用光.也就是說，下一次你調用這個查詢方法，就可能會在一個時間與其他的parallel stream同時運行，而導致第二個任務的性能大大受損。或者說，例如你在這個功能里是用來快速返回調用的第三方api的,而在其他的功能里是用於一些簡單的數據並行計算的,但是假如你先調用了這個功能,同一時間之后調用計算的函數,那么這里forkjionPool的實現會讓你計算的函數大打折扣.
不過也不要急着去吐槽ForkJoinPool的實現,在不同的情況下你可以給它一個ManagedBlocker實例並且確保它知道在一個阻塞調用中應該什么時候去抵消掉卡住的workers.現在有意思的一點是，在一個parallel stream處理中並不一定是阻塞調用會拖延程序的性能。任何被用於映射在一個集合上的長時間運行的函數都會產生同樣的問題.
正如我們上面那個列子的情況分析得知,lambda的執行並不是瞬間完成的,所有使用parallel streams的程序都有可能成為阻塞程序的源頭,並且在執行過程中程序中的其他部分將無法訪問這些workers,這意味着任何依賴parallel streams的程序在什么別的東西占用着common ForkJoinPool時將會變得不可預知並且暗藏危機.
怎么正確使用parallelStream
如果你正在寫一個其他地方都是單線程的程序並且准確地知道什么時候你應該要使用parallel streams，這樣的話你可能會覺得這個問題有一點膚淺。然而，我們很多人是在處理web應用、各種不同的框架以及重量級應用服務。一個服務器是怎樣被設計成一個可以支持多種獨立應用的主機的？誰知道呢，給你一個可以並行的卻不能控制輸入的parallel stream.
很抱歉,請原諒我用的標注[怎么正確使用parallelStream],因為目前為止我也沒有發現一個好的方式來讓我真正的正確使用parallelStream.下面的網上寫的兩種方式:
一種方式是限制ForkJoinPool提供的並行數。可以通過使用-Djava.util.concurrent.ForkJoinPool.common.parallelism=1 來限制線程池的大小為1。不再從並行化中得到好處可以杜絕錯誤的使用它(其實這個方式還是有點搞笑的,既然這樣搞那我還不如不去使用並行流)。
另一種方式就是，一個被稱為工作區的可以讓ForkJoinPool平行放置的 parallelStream() 實現。不幸的是現在的JDK還沒有實現。
Parallel streams 是無法預測的，而且想要正確地使用它有些棘手。幾乎任何parallel streams的使用都會影響程序中無關部分的性能，而且是一種無法預測的方式。。但是在調用stream.parallel() 或者parallelStream()時候在我的代碼里之前我仍然會重新審視一遍他給我的程序究竟會帶來什么問題,他能有多大的提升,是否有使用他的意義.
stream or parallelStream？
上面我們也看到了parallelStream所帶來的隱患和好處,那么,在從stream和parallelStream方法中進行選擇時,我們可以考慮以下幾個問題：
1. 是否需要並行？
2. 任務之間是否是獨立的？是否會引起任何競態條件？
3. 結果是否取決於任務的調用順序？
對於問題1，在回答這個問題之前，你需要弄清楚你要解決的問題是什么，數據量有多大，計算的特點是什么？並不是所有的問題都適合使用並發程序來求解，比如當數據量不大時，順序執行往往比並行執行更快。畢竟，准備線程池和其它相關資源也是需要時間的。但是，當任務涉及到I/O操作並且任務之間不互相依賴時，那么並行化就是一個不錯的選擇。通常而言，將這類程序並行化之后，執行速度會提升好幾個等級。
對於問題2，如果任務之間是獨立的，並且代碼中不涉及到對同一個對象的某個狀態或者某個變量的更新操作，那么就表明代碼是可以被並行化的。
對於問題3，由於在並行環境中任務的執行順序是不確定的，因此對於依賴於順序的任務而言，並行化也許不能給出正確的結果。