java8并行流怎么應用

這篇文章主要介紹了java8并行流怎么應用的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇java8并行流怎么應用文章都會有所收獲，下面我們一起來看看吧。

并行流

在java7之前,處理并行數據非常麻煩.
第一:你得明確的把包含的數據結構分成若干子部分.
第二:你要給每個子部分分配獨立的線程.
第三:你需要在恰當的時候對他們進行同步,來避免不希望出現的競爭條件,等待所有線程完成,最后把這些結果合并起來.

并行流就是把一個內容分成多個數據塊,并用不同線程分別處理每個數據塊的流.

這樣一來,你就可以把給定的工作負荷自動分配給多個處理器內核,讓他們都忙起來.

假設你需要寫一個方法，接受數字n作為參數，并返回從1到給定參數的所有數字的和。一個
直接（也許有點土）的方法是生成一個無窮大的數字流，把它限制到給定的數目，然后用對兩個
數字求和的 BinaryOperator 來歸約這個流，如下所示：

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

用更為傳統的Java術語來說，這段代碼與下面的迭代等價

//傳統迭代
    public static Long iterativeSum(Long n){
        Long result = 0L;
        for (Long i=1L;i<n;i++){
            result = result+i;
        }
        return

這似乎是利用并行流的好機會,特別是n很大的時候,那該怎樣做呢?

你要對結果變量進行同步嗎？用多少個線程呢？誰負責生成數呢？誰來做加法呢？

其實根本不必擔心,并行流已經幫我們做完了這些令人頭疼的工作

將順序流轉換為并行流

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

在現實中，對順序流調用 parallel 方法并不意味著流本身有任何實際的變化。它
在內部實際上就是設了一個 boolean 標志，表示你想讓調用 parallel 之后進行的所有操作都并
行執行.類似地，你只需要對并行流調用 sequential 方法就可以把它變成順序流.

測量流性能

我們說并行求和的方法應該比順序迭代的方法更好.但在軟件工程上, 靠猜絕對不是什么好辦法,有時候經驗也靠不住.
你應該始終遵循三個黃金規則;測量,測量,再測量.

為了簡化測量,我們寫個方法,專門用來測試 ParallelStreams類里的三個求和方法: sequentialSum iterativeSum, parallelSum.

public Long measureSumPref(Function<Long, Long> addr, Long n) {
        long fastest = Long.MAX_VALUE;
        for (int i = 0; i < 10; i++) {
            Long start = System.nanoTime();
            Long sum = addr.apply(n);
            Long druation = (System.nanoTime() - start)/1000000;
            if (druation < fastest) {
                fastest = druation;
            }
        }
        return

這個方法會接收一個函數和一個Long類型參數.它會對傳給方法的參數應用函數10次,記錄每次執行的時間.

下面是測試結果

//順序
    @Test
    public void test4() {
        Long fast = measureSumPref(ParallelStreams::sequentialSum, 1000 * 10000L);
        System.out.println("sequentialSum= " + fast);//398毫秒


    }

    //迭代
    @Test
    public void test5() {
        Long fast = measureSumPref(ParallelStreams::iterativeSum, 1000 * 10000L);
        System.out.println("iterativeSum= "+ fast);//153毫秒


    }
    //并行
    @Test
    public void test6(){
        Long fast = measureSumPref(ParallelStreams::parallelSum, 1000 * 10000L);
        System.out.println("parallelSum= "+fast);//1309毫秒

看到結果,我們發現并行流操作相當令我們失望.

求和方法的并行版本比順序版本要慢很多!!! 其實對這個意外的結果,有兩方面的原因:

• 一:iterate 生成的是裝箱對象,必須拆箱成數字才能求和.

• 二:我們很難把iterate分成多個獨立的塊來執行.

對于第二個問題,很有意思,我們直覺上可能是這樣運行的,如圖:

但是,iterate 很難分割成能夠獨立執行的小塊，因為每次應用這個函數都要依賴前一次應用的結果

也就是說,整張數字表在歸納過程開始時還沒準備好,因為Stream在遇到終端操作才會開始執行,因而無法有效的把流劃分為小塊進行處理.
把流標記為并行,其實是給順序處理增加了開銷,它還要把每次求和操作的結果分到一個不同的線程上.

這就說明了并行編程肯能很復雜,如果用得不對（比如采用了一個不易并行化的操作，如 iterate ），它甚至可能讓程序的整體性能更差.
所以在調用那個看似神奇的 parallel 操作時，了解背后到底發生了什么是很有必要的。

并行流使用注意事項:使用更有針對性的方法

對于上面那種出人意料的結果,我們萬不可把鍋退給并行流,其實仔細分析,不難發現,這是我們使用了不恰當的的數據結構導致的.

對于上面的并行處理操作,我們可做如下改進.在之前的文章中,我們介紹過一個叫LongStream的流.這個流有個專門針對Long型的方法

• LongStream.rangeClosed 直接產生原始類型的long數字,沒有裝箱拆箱的開銷.

• LongStream.rangeClosed 會生成數字范圍,很容易查分為獨立的小塊.

LongStream和Stream一樣都繼承了BaseStream

public interface LongStream extends BaseStream<Long, LongStream> {...}

public interface Stream<T> extends BaseStream<T, Stream<T>> {...}

這兩個流的用法基本完全相同,唯一的不同相比從名字就能看出來,LongStream 指明了流類型為Long,類似的還有,IntStream,DoubleStream等

我們改進代碼如下:

//順序流改進版 LongStream.rangeClosed
    public static Long sequentialSum2(Long n) {
        return LongStream.rangeClosed(1, n)
                .reduce(0L,Long::sum);

    }

 //并行流改進版
    public static Long paraparallelSum2(Long n) {
        return LongStream.rangeClosed(1, n)
                .parallel()
                .reduce(0L,Long::sum);

    }

然后再次進行測量

//順序流(改進版)
    @Test
    public void test7(){
        Long fast = measureSumPref(ParallelStreams::sequentialSum2, 1000 * 10000L);
        System.out.println("順序流(改進版)="+fast);//56毫秒------改進之前:398毫秒

    }



    //并行流(改進版)
    @Test
    public void test8(){
        Long fast = measureSumPref(ParallelStreams::paraparallelSum2, 1000 * 10000L);
        System.out.println("并行流(改進版)="+fast);//14毫秒--------改進之前:1309毫秒

由此結果可得出結論:

• 使用LongStream比iterate效率提高 710%

• 在上面基礎上使用 并行流 比 順序流 效率提高 400%

可見:選擇適當的數據結構往往比并行算法正重要,使用正確的數據結構后再選擇并行算法能保證最佳的性能.

盡管如此,我們也必須知道,并行化不是沒有代價的.并行化本身需要對流做遞歸劃分,把每個子流的歸納操作分配到不同的線程,然后把這些操作的結果合并成一個值.
但在多個內核之間移動數據的代價也可能比你想的要大,所以在使用并行操作很重要的一點就是要保證并行執行的工作時間要比數據在內核之前移動的時間要長.

在使用并行Stream加速代碼之前,你必須確保用的對,如果用錯了,算得快就毫無意義了.讓我們看一個常見的陷阱.

高效使用并行流

• 如果有疑問，測量。把順序流轉成并行流輕而易舉，但卻不一定是好事

• 留意裝箱。自動裝箱和拆箱操作會大大降低性能。Java 8中有原始類型流（ IntStream 、
  LongStream 、 DoubleStream ）來避免這種操作，但凡有可能都應該用這些流

• 有些操作本身在并行流上的性能就比順序流差。特別是 limit 和 findFirst 等依賴于元
  素順序的操作，它們在并行流上執行的代價非常大。例如， findAny 會比 findFirst 性
  能好，因為它不一定要按順序來執行。

• 還要考慮流的操作流水線的總計算成本。設N是要處理的元素的總數，Q是一個元素通過
  流水線的大致處理成本，則N*Q就是這個對成本的一個粗略的定性估計。Q值較高就意味
  著使用并行流時性能好的可能性比較大。

• 對于較小的數據量，選擇并行流幾乎從來都不是一個好的決定。并行處理少數幾個元素
  的好處還抵不上并行化造成的額外開銷

• 要考慮流背后的數據結構是否易于分解。例如， ArrayList 的拆分效率比 LinkedList
  高得多，因為前者用不著遍歷就可以平均拆分，而后者則必須遍歷

• 還要考慮終端操作中合并步驟的代價是大是小（例如 Collector 中的 combiner 方法）

需要強調的是:并行流背后使用的基礎架構是java7引入的分支/合并框架.我們想要正確高效的使用并行流,了解它的內部原理至關重要.

參考:分支/合并框架

Spliterator

Spliterator 是Java 8中加入的另一個新接口；這個名字代表“可分迭代器”（splitable
iterator）。和 Iterator 一樣， Spliterator 也用于遍歷數據源中的元素，但它是為了并行執行
而設計的。雖然在實踐中可能用不著自己開發 Spliterator ，但了解一下它的實現方式會讓你
對并行流的工作原理有更深入的了解。Java 8已經為集合框架中包含的所有數據結構提供了一個
默認的 Spliterator 實現。

關于“java8并行流怎么應用”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“java8并行流怎么應用”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注億速云行業資訊頻道。