緒論

之前的幾章中，我們已經(jīng)看到了新的Stream接口可以讓你以聲明性方式處理數(shù)據(jù)集。我們還解釋了將外部迭代換為內(nèi)部迭代能夠讓原生Java庫(kù)控制流元素的處理。這種方法讓Java程序員無(wú)需顯示實(shí)現(xiàn)優(yōu)化來(lái)為數(shù)據(jù)集的處理加速。到目前為止，最重要的好處是可以對(duì)這些集合執(zhí)行操作流水線，能夠自動(dòng)利用計(jì)算機(jī)上的多個(gè)內(nèi)核。

在本篇中，你將了解Stream接口如何讓你不用太費(fèi)力氣就能對(duì)數(shù)據(jù)執(zhí)行并行操作。它允許你聲明性的將順序流變?yōu)椴⑿辛鳌４送?，你講看到Java是如何變戲法的，或者更實(shí)際地來(lái)說(shuō)，流是如何在幕后應(yīng)用java7引入的分支/合并框架的。你還會(huì)發(fā)現(xiàn)，了解并行流內(nèi)部是如何工作的很重要，因?yàn)槿绻愫鲆曔@一方面，就可能因誤用而得到意外(很可能是錯(cuò)的)的結(jié)果。

為此，會(huì)特別顯示，在并行處理數(shù)據(jù)塊之前，并行流被劃分為數(shù)據(jù)塊的方式在某些情況下恰恰是這些錯(cuò)誤且無(wú)法解釋的結(jié)果的根源。因此，你將會(huì)學(xué)習(xí)到如何通過(guò)實(shí)現(xiàn)和使用你自己的Spliterator來(lái)控制這個(gè)劃分過(guò)程。

并行流

可以通過(guò)對(duì)收集源調(diào)用parallelStream方法來(lái)把集合轉(zhuǎn)換為并行流。并行流就是一個(gè)把內(nèi)容分成多個(gè)數(shù)據(jù)塊，并用不同的線程分別處理每個(gè)數(shù)據(jù)塊的流。這樣一來(lái)，你就可以自動(dòng)把給定操作的工作負(fù)荷分配給多核處理器的所有內(nèi)核，讓他們都忙起來(lái)。

假設(shè)要寫一個(gè)方法，接受數(shù)字n作為參數(shù)，并返回從1到給定參數(shù)的的所有數(shù)字的和。一個(gè)直接的方法是生成一個(gè)無(wú)窮大的數(shù)字流，把它限制到給定的數(shù)目，然后用對(duì)兩個(gè)數(shù)字求和的BinaryOperater來(lái)歸約這個(gè)流

public static long sequentialSum(long n){
    return Stream.iterate(1L,i ->i + 1)
    .limit(n).reduce(0L,long :: sum);
}

這似乎是利用并行處理的好機(jī)會(huì)，特別是n很大的時(shí)候。那怎么入手呢？你要對(duì)結(jié)果變量進(jìn)行同步嗎？用多少個(gè)線程呢？誰(shuí)負(fù)責(zé)生成數(shù)呢？誰(shuí)來(lái)做加法呢？

根本不用擔(dān)心。用并行流的話，這問(wèn)題就簡(jiǎn)單多了。

將順序流轉(zhuǎn)化為并行流

可以把流轉(zhuǎn)化成并行流，從而讓前面的函數(shù)歸約過(guò)程(也就是求和)并行運(yùn)行——對(duì)順序流調(diào)用parallel方法：

public static long parallelSum(long n){
    return Stream.iterate(1L,i -> i +1)
    .limit(n)
    .parallel()
    .reduce(0L,Long::sum);
}

上面的代碼的不同之處在于Stream在內(nèi)部分成了幾塊。因此可以對(duì)不同的快獨(dú)立并行進(jìn)行歸納操作。

image

請(qǐng)注意，在現(xiàn)實(shí)中，對(duì)順序流調(diào)用parallel方法并不意味著流本身有任何實(shí)際的變化。它在內(nèi)部實(shí)際上就是設(shè)了一個(gè)boolean標(biāo)志，表示你想讓調(diào)用parallel之后進(jìn)行的所有操作都并行執(zhí)行。類似的，你只需要對(duì)并行流調(diào)用sequential方法就可以刻把它變成順序流。請(qǐng)注意，你可能以為吧這兩個(gè)方法都結(jié)合起來(lái)，就可以更細(xì)化地控制在遍歷流時(shí)那些操作要并行執(zhí)行，哪些要順序執(zhí)行。例如：

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

但是最后一次parallel或者sequential調(diào)用會(huì)影響整個(gè)流水線。在本例中，流水線會(huì)并行執(zhí)行，因?yàn)樽詈笳{(diào)用的是它。

那么，調(diào)用parallel方法，你可能會(huì)想，并行流用的線程是從哪兒 來(lái)的呢？有多少個(gè)？怎么定義這個(gè)過(guò)程？
并行流內(nèi)部使用了默認(rèn)的ForkJoinPool，它默認(rèn)的是線程數(shù)量就是處理器的數(shù)量，這個(gè)值是Runtime.getRuntime().availableProcessors()得到的。

但是你可一個(gè)通過(guò)系統(tǒng)屬性來(lái)改變。System.setProperty();這是一個(gè)全局設(shè)置，因此它將影響代碼中所有的的并行流。反過(guò)來(lái)說(shuō)，目前還無(wú)法專為某個(gè)并行流指定這個(gè)值。一般而言，讓ForkJoinPool的大小等于處理器數(shù)量是個(gè)不錯(cuò)的默認(rèn)值，強(qiáng)烈不建議修改它。

測(cè)量流性能

我們聲稱并行求和方法應(yīng)該比順序求和迭代方法性能好。然而在軟件工程上，靠猜絕對(duì)不是什么好辦法！特別是在優(yōu)化性能時(shí)，要測(cè)量，測(cè)量，再測(cè)量。
例子：測(cè)量對(duì)前n個(gè)自然數(shù)求和的函數(shù)的性能

public long measureSumPerf(Function<Long ,Long> adder,long n){
    long fastest = Long.MAX_VALUE;
    for(int i= 0;i<10;i++){
        long start = System.nanoTime();
        long sum = adder.apply(n);
        long duration = (System.nanoTime()-start)/1000000;
        System.out.println("Result"+sum);
        if(duration<fastest) fastest = duration;
    }
    return fastest;
}

現(xiàn)在就可以把先前開發(fā)的所有方法都放進(jìn)了一個(gè)名為ParallelStreams的類，你就可以用這個(gè)框架來(lái)測(cè)試書序加法器函數(shù)對(duì)前已前往個(gè)自然數(shù)求和要多久：

System.out.println(meadureSumPerf(ParallelStreams::sequentialSum,10000000))

//結(jié)果：97

用傳統(tǒng)for循環(huán)的迭代版本執(zhí)行起來(lái)應(yīng)該會(huì)快很多，因?yàn)樗鼮榈讓樱匾氖遣恍枰獙?duì)原始類型做任何裝箱和拆箱的工作。

System.out.println(measureSumPerf(ParallelStreams::iterativeSum,10000000));
//結(jié)果： 2

現(xiàn)在我們對(duì)函數(shù)的并行版本做測(cè)試：

System.out.println(measureSumPerf(ParallelStreams::parallelSum,10000000));
//結(jié)果：164

令人相當(dāng)?shù)氖?，求和方法的并行版本比順序?zhí)行版本要慢的多。這里有兩個(gè)實(shí)際問(wèn)題：

1. iterate生成的是裝箱的對(duì)象，必須拆箱成數(shù)字才能求和；
2. 我們很難吧iterate分成多個(gè)獨(dú)立塊來(lái)并行執(zhí)行。

iterate很難分割成能夠獨(dú)立執(zhí)行的小塊，因?yàn)槊看螒?yīng)用這個(gè)函數(shù)都要依賴前一次用用的結(jié)果：

image

這意味著，在這個(gè)特定的情況下，歸納進(jìn)程不是像將順序流轉(zhuǎn)化成并行流那樣進(jìn)行的；整張數(shù)字列表在歸納過(guò)程開始時(shí)沒有準(zhǔn)備好，因而無(wú)法有效地把流劃分為小塊來(lái)并行處理。把流標(biāo)記成并行，你其實(shí)是個(gè)順序處理增加了開銷，他還要八二每次求和操作分到一個(gè)不同的線程上。

這就說(shuō)明了并行編程可能很復(fù)雜，有時(shí)候甚至有點(diǎn)違反直覺。如果用的不對(duì)(比如才用了一個(gè)不易并行化的操作，如iterate)，它甚至可能讓程序的整體性能更差，所以在調(diào)用那個(gè)看似神奇的aprallel操作時(shí)，了解背后到底發(fā)生了什么是很有必要的。

使用更有針對(duì)性的方法

那到底要怎么利用多核處理器，用流來(lái)高效地并行求和呢？我們?cè)诘?章中討論了一個(gè)叫LongStream.rangeClosed的方法。這個(gè)方法與iterate相比有兩個(gè)優(yōu)點(diǎn)。

• LongStream.rangeClosed直接產(chǎn)生原始類型的long數(shù)字，沒有裝箱拆箱的操作
• LongStream.rangeClosed會(huì)生成數(shù)字范圍，很容易拆分為獨(dú)立的小塊。列如，范圍1_20可分為15，6_10，1115，16~20。
  先看一下它用于順序流時(shí)的性能如何，看看拆箱的開銷：

public static long rangedSum(long n){
    return LongStream.rangeClosed(1,n).reduce(0L,Long::sum);
}
//輸出 17

這個(gè)數(shù)值流比前面那個(gè)用iterate工廠方法生成數(shù)字的順序執(zhí)行版本要快得多，因?yàn)閿?shù)值流避免了非針對(duì)想流那些沒必要的自動(dòng)裝箱和拆箱的操作。由此可見，選擇適當(dāng)?shù)臄?shù)據(jù)結(jié)構(gòu)往往比并行化算法更為有效。使用并行的效果呢？

public static long rangedSum(long n){
    return LongStream.rangeClosed(1,n).parallel().reduce(0L,Long::sum);
}
//輸出 1

終于有了一個(gè)比順序執(zhí)行更快的并行歸納。這也表明，使用正確的數(shù)據(jù)結(jié)構(gòu)然后使其并行工作能夠保證最佳的性能。

盡管如此，請(qǐng)記住，并行化并不是沒有代價(jià)的。并行化過(guò)程本身需要對(duì)流做遞歸劃分，把每個(gè)子流的歸納操作分配到不同的線程，然后把這些操作的結(jié)果合并成一個(gè)值。但在多個(gè)內(nèi)核之間移動(dòng)數(shù)據(jù)的代價(jià)也可能比你想要的要大，所以很重要的一點(diǎn)是要辦證在內(nèi)核中并行執(zhí)行工作的時(shí)間比在內(nèi)核之間傳輸數(shù)據(jù)的時(shí)間長(zhǎng)。

高效使用并行流

• 如果有疑問(wèn)，測(cè)量。把順序轉(zhuǎn)成并行流輕而易舉，但卻比一定是好事。因?yàn)椴⑿辛鞑⒉豢偸潜软樞蚩臁?/li>
• 留意裝箱。自動(dòng)裝箱和拆箱操作會(huì)大大降低性能。java8中有原始類型流(IntStream,LongStream,DoubleStream)來(lái)避免這種操作，但凡有可能都要用這些流。
• 有些操作本身在并行流上的性能就比順序流查。特別是limit和findFirst等依賴于元素順序的操作，他們?cè)诓⑿辛魃蠄?zhí)行的代價(jià)非常大。findAny會(huì)比f(wàn)indFirst性能好，因?yàn)樗灰欢ㄒ错樞驁?zhí)行。
• 對(duì)于較小的數(shù)據(jù)量，選擇并行流幾乎從來(lái)都不是一個(gè)好的決定。并行處理少數(shù)幾個(gè)元素的好處還抵不上并行化造成額外的開銷。

• 看流的內(nèi)容是否可拆分

  
  
  image
  
  

  流背后使用的基礎(chǔ)架構(gòu)是java7中引入的分支/合并框架。并行匯總的實(shí)例證明了要想正確使用并行流，了解它的內(nèi)部原理至關(guān)重要。

分支/合并框架

分支/合并框架的目的是以遞歸方式將可以并行的任務(wù)拆分成更小的任務(wù)，然后將每個(gè)子任務(wù)的結(jié)果合并起來(lái)生成整體結(jié)果。它是ExecutorService接口的一個(gè)實(shí)現(xiàn)，它把子任務(wù)分配給線程池(稱為ForkJoinPool)中的工作線程。