通過實例解析java8中的parallelStream

這篇文章主要介紹了通過實例解析java8中的parallelStream,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下

about Stream

什么是流？

Stream是java8中新增加的一個特性,被java猿統(tǒng)稱為流.

Stream 不是集合元素，它不是數(shù)據(jù)結(jié)構(gòu)并不保存數(shù)據(jù)，它是有關(guān)算法和計算的，它更像一個高級版本的 Iterator。原始版本的 Iterator，用戶只能顯式地一個一個遍歷元素并對其執(zhí)行某些操作；高級版本的 Stream，用戶只要給出需要對其包含的元素執(zhí)行什么操作，比如 “過濾掉長度大于 10 的字符串”、“獲取每個字符串的首字母”等，Stream 會隱式地在內(nèi)部進(jìn)行遍歷，做出相應(yīng)的數(shù)據(jù)轉(zhuǎn)換。

Stream 就如同一個迭代器（Iterator），單向，不可往復(fù)，數(shù)據(jù)只能遍歷一次，遍歷過一次后即用盡了，就好比流水從面前流過，一去不復(fù)返。

而和迭代器又不同的是，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 的另外一大特點是，數(shù)據(jù)源本身可以是無限的。

parallelStream是什么

parallelStream其實就是一個并行執(zhí)行的流.它通過默認(rèn)的ForkJoinPool,可能提高你的多線程任務(wù)的速度.

parallelStream的作用
Stream具有平行處理能力，處理的過程會分而治之，也就是將一個大任務(wù)切分成多個小任務(wù)，這表示每個任務(wù)都是一個操作，因此像以下的程式片段：

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

你得到的展示順序不一定會是1、2、3、4、5、6、7、8、9，而可能是任意的順序，就forEach()這個操作來講，如果平行處理時，希望最后順序是按照原來Stream的數(shù)據(jù)順序，那可以調(diào)用forEachOrdered()。例如：

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

注意:如果forEachOrdered()中間有其他如filter()的中介操作，會試著平行化處理，然后最終forEachOrdered()會以原數(shù)據(jù)順序處理，因此，使用forEachOrdered()這類的有序處理,可能會（或完全失去）失去平行化的一些優(yōu)勢，實際上中介操作亦有可能如此，例如sorted()方法。

parallelStream背后的男人:ForkJoinPool

要想深入的研究parallelStream之前,那么我們必須先了解ForkJoin框架和ForkJoinPool.本文旨在parallelStream,但因為兩種關(guān)系甚密,故在此簡單介紹一下ForkJoinPool,如有興趣可以更深入的去了解下ForkJoin***(當(dāng)然,如果你想真正的搞透parallelStream,那么你依然需要先搞透ForkJoinPool).*

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

ForkJoinPool主要用來使用分治法(Divide-and-Conquer Algorithm)來解決問題。典型的應(yīng)用比如快速排序算法。這里的要點在于，F(xiàn)orkJoinPool需要使用相對少的線程來處理大量的任務(wù)。比如要對1000萬個數(shù)據(jù)進(jìn)行排序，那么會將這個任務(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(jìn)行排序。那么到最后，所有的任務(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萬個線程，顯然這是不可行的。

工作竊取算法

forkjoin最核心的地方就是利用了現(xiàn)代硬件設(shè)備多核,在一個操作時候會有空閑的cpu,那么如何利用好這個空閑的cpu就成了提高性能的關(guān)鍵,而這里我們要提到的工作竊?。╳ork-stealing）算法就是整個forkjion框架的核心理念,工作竊?。╳ork-stealing）算法是指某個線程從其他隊列里竊取任務(wù)來執(zhí)行。

那么為什么需要使用工作竊取算法呢？

假如我們需要做一個比較大的任務(wù)，我們可以把這個任務(wù)分割為若干互不依賴的子任務(wù)，為了減少線程間的競爭，于是把這些子任務(wù)分別放到不同的隊列里，并為每個隊列創(chuàng)建一個單獨(dú)的線程來執(zhí)行隊列里的任務(wù)，線程和隊列一一對應(yīng)，比如A線程負(fù)責(zé)處理A隊列里的任務(wù)。但是有的線程會先把自己隊列里的任務(wù)干完，而其他線程對應(yīng)的隊列里還有任務(wù)等待處理。干完活的線程與其等著，不如去幫其他線程干活，于是它就去其他線程的隊列里竊取一個任務(wù)來執(zhí)行。而在這時它們會訪問同一個隊列，所以為了減少竊取任務(wù)線程和被竊取任務(wù)線程之間的競爭，通常會使用雙端隊列，被竊取任務(wù)線程永遠(yuǎn)從雙端隊列的頭部拿任務(wù)執(zhí)行，而竊取任務(wù)的線程永遠(yuǎn)從雙端隊列的尾部拿任務(wù)執(zhí)行。

工作竊取算法的優(yōu)點是充分利用線程進(jìn)行并行計算，并減少了線程間的競爭，其缺點是在某些情況下還是存在競爭，比如雙端隊列里只有一個任務(wù)時。并且消耗了更多的系統(tǒng)資源，比如創(chuàng)建多個線程和多個雙端隊列。

用看forkjion的眼光來看ParallelStreams

上文中已經(jīng)提到了在Java 8引入了自動并行化的概念。它能夠讓一部分Java代碼自動地以并行的方式執(zhí)行，也就是我們使用了ForkJoinPool的ParallelStream。

Java 8為ForkJoinPool添加了一個通用線程池，這個線程池用來處理那些沒有被顯式提交到任何線程池的任務(wù)。它是ForkJoinPool類型上的一個靜態(tài)元素，它擁有的默認(rèn)線程數(shù)量等于運(yùn)行計算機(jī)上的處理器數(shù)量。當(dāng)調(diào)用Arrays類上添加的新方法時，自動并行化就會發(fā)生。比如用來排序一個數(shù)組的并行快速排序，用來對一個數(shù)組中的元素進(jìn)行并行遍歷。自動并行化也被運(yùn)用在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中的通用線程池處理。以上的并行計算邏輯當(dāng)然也可以使用ThreadPoolExecutor完成，但是就代碼的可讀性和代碼量而言，使用ForkJoinPool明顯更勝一籌。

對于ForkJoinPool通用線程池的線程數(shù)量，通常使用默認(rèn)值就可以了，即運(yùn)行時計算機(jī)的處理器數(shù)量。我這里提供了一個示例的代碼讓你了解jvm所使用的ForkJoinPool的線程數(shù)量, 你可以可以通過設(shè)置系統(tǒng)屬性：-Djava.util.concurrent.ForkJoinPool.common.parallelism=N （N為線程數(shù)量）,來調(diào)整ForkJoinPool的線程數(shù)量,可以嘗試調(diào)整成不同的參數(shù)來觀察每次的輸出結(jié)果:

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!");
    // 構(gòu)造一個10000個元素的集合
    List<Integer> list = new ArrayList<>();
    for (int i = 0; i < 10000; i++) {
      list.add(i);
    }
    // 統(tǒng)計并行執(zhí)行l(wèi)ist的線程
    Set<Thread> threadSet = new CopyOnWriteArraySet<>();
    // 并行執(zhí)行
    list.parallelStream().forEach(integer -> {
      Thread thread = Thread.currentThread();
      // System.out.println(thread);
      // 統(tǒng)計并行執(zhí)行l(wèi)ist的線程
      threadSet.add(thread);
    });
    System.out.println("threadSet一共有" + threadSet.size() + "個線程");
    System.out.println("系統(tǒng)一個有"+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("系統(tǒng)一個有"+Runtime.getRuntime().availableProcessors()+"個cpu");
  }
}

出現(xiàn)這種現(xiàn)象的原因是，forEach方法用了一些小把戲。它會將執(zhí)行forEach本身的線程也作為線程池中的一個工作線程。因此，即使將ForkJoinPool的通用線程池的線程數(shù)量設(shè)置為1，實際上也會有2個工作線程。因此在使用forEach的時候，線程數(shù)為1的ForkJoinPool通用線程池和線程數(shù)為2的ThreadPoolExecutor是等價的。

所以當(dāng)ForkJoinPool通用線程池實際需要4個工作線程時，可以將它設(shè)置成3，那么在運(yùn)行時可用的工作線程就是4了。

小結(jié):

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

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

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

ParallelStreams 的陷阱

上文中我們已經(jīng)看到了ParallelStream他強(qiáng)大無比的特性,但這里我們就講告訴你ParallelStreams不是萬金油,而是一把雙刃劍,如果錯誤的使用反倒可能傷人傷己.

以下是一個我們項目里使用 parallel streams 的很常見的情況。在這個例子中，我們想同時調(diào)用不同地址的api中并且獲得第一個返回的結(jié)果。

  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自己實現(xiàn)的這個功能,但是jdk8的朋友基本都會用上面的實現(xiàn)方式,那么自信深究一下究竟自己用future實現(xiàn)的這個功能和利用jdk8的parallelStream來實現(xiàn)這個功能有什么不同點呢?坑又在哪里呢？

讓我們細(xì)思思考一下整個功能究竟是如何運(yùn)轉(zhuǎn)的。首先我們的集合元素engines 由ParallelStreams并行的去進(jìn)行map操作(ParallelStreams使用JVM默認(rèn)的forkJoin框架的線程池由當(dāng)前線程去執(zhí)行并行操作).

然而,這里需要注意的一地方是我們在調(diào)用第三方的api請求是一個響應(yīng)略慢而且會阻塞操作的一個過程。所以在某時刻所有線程都會調(diào)用 get() 方法并且在那里等待結(jié)果返回.

再回過頭仔細(xì)思考一下這個功能的實現(xiàn)過程是我們一開始想要的嗎？我們是在同一時間等待所有的結(jié)果,而不是遍歷這個列表按順序等待每個回答.然而，由于ForkJoinPool workders的存在，這樣平行的等待相對于使用主線程的等待會產(chǎn)生的一種副作用.

現(xiàn)在ForkJoin pool (關(guān)于forkjion的更多實現(xiàn)你可以去搜索引擎中去看一下他的具體實現(xiàn)方式) 的實現(xiàn)是: 它并不會因為產(chǎn)生了新的workers而抵消掉阻塞的workers。那么在某個時間所有 ForkJoinPool.common() 的線程都會被用光.也就是說，下一次你調(diào)用這個查詢方法，就可能會在一個時間與其他的parallel stream同時運(yùn)行，而導(dǎo)致第二個任務(wù)的性能大大受損?；蛘哒f，例如你在這個功能里是用來快速返回調(diào)用的第三方api的,而在其他的功能里是用于一些簡單的數(shù)據(jù)并行計算的,但是假如你先調(diào)用了這個功能,同一時間之后調(diào)用計算的函數(shù),那么這里forkjionPool的實現(xiàn)會讓你計算的函數(shù)大打折扣.

不過也不要急著去吐槽ForkJoinPool的實現(xiàn),在不同的情況下你可以給它一個ManagedBlocker實例并且確保它知道在一個阻塞調(diào)用中應(yīng)該什么時候去抵消掉卡住的workers.現(xiàn)在有意思的一點是，在一個parallel stream處理中并不一定是阻塞調(diào)用會拖延程序的性能。任何被用于映射在一個集合上的長時間運(yùn)行的函數(shù)都會產(chǎn)生同樣的問題.

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

怎么正確使用parallelStream

如果你正在寫一個其他地方都是單線程的程序并且準(zhǔn)確地知道什么時候你應(yīng)該要使用parallel streams，這樣的話你可能會覺得這個問題有一點膚淺。然而，我們很多人是在處理web應(yīng)用、各種不同的框架以及重量級應(yīng)用服務(wù)。一個服務(wù)器是怎樣被設(shè)計成一個可以支持多種獨(dú)立應(yīng)用的主機(jī)的？誰知道呢，給你一個可以并行的卻不能控制輸入的parallel stream.

很抱歉,請原諒我用的標(biāo)注[怎么正確使用parallelStream],因為目前為止我也沒有發(fā)現(xiàn)一個好的方式來讓我真正的正確使用parallelStream.下面的網(wǎng)上寫的兩種方式:

一種方式是限制ForkJoinPool提供的并行數(shù)?？梢酝ㄟ^使用-Djava.util.concurrent.ForkJoinPool.common.parallelism=1 來限制線程池的大小為1。不再從并行化中得到好處可以杜絕錯誤的使用它(其實這個方式還是有點搞笑的,既然這樣搞那我還不如不去使用并行流)。

另一種方式就是，一個被稱為工作區(qū)的可以讓ForkJoinPool平行放置的 parallelStream() 實現(xiàn)。不幸的是現(xiàn)在的JDK還沒有實現(xiàn)。

Parallel streams 是無法預(yù)測的，而且想要正確地使用它有些棘手。幾乎任何parallel streams的使用都會影響程序中無關(guān)部分的性能，而且是一種無法預(yù)測的方式。。但是在調(diào)用stream.parallel() 或者parallelStream()時候在我的代碼里之前我仍然會重新審視一遍他給我的程序究竟會帶來什么問題,他能有多大的提升,是否有使用他的意義.

stream or parallelStream？

上面我們也看到了parallelStream所帶來的隱患和好處,那么,在從stream和parallelStream方法中進(jìn)行選擇時,我們可以考慮以下幾個問題：

1. 是否需要并行？

2. 任務(wù)之間是否是獨(dú)立的？是否會引起任何競態(tài)條件？

3. 結(jié)果是否取決于任務(wù)的調(diào)用順序？

對于問題1，在回答這個問題之前，你需要弄清楚你要解決的問題是什么，數(shù)據(jù)量有多大，計算的特點是什么？并不是所有的問題都適合使用并發(fā)程序來求解，比如當(dāng)數(shù)據(jù)量不大時，順序執(zhí)行往往比并行執(zhí)行更快。畢竟，準(zhǔn)備線程池和其它相關(guān)資源也是需要時間的。但是，當(dāng)任務(wù)涉及到I/O操作并且任務(wù)之間不互相依賴時，那么并行化就是一個不錯的選擇。通常而言，將這類程序并行化之后，執(zhí)行速度會提升好幾個等級。

對于問題2，如果任務(wù)之間是獨(dú)立的，并且代碼中不涉及到對同一個對象的某個狀態(tài)或者某個變量的更新操作，那么就表明代碼是可以被并行化的。

對于問題3，由于在并行環(huán)境中任務(wù)的執(zhí)行順序是不確定的，因此對于依賴于順序的任務(wù)而言，并行化也許不能給出正確的結(jié)果。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持億速云。