java并發(fā)編程學(xué)習(xí)6--并行流

mgckid 發(fā)布于2019-08-15 13:26 / 631人閱讀

摘要：類(lèi)似的你可以用將并行流變?yōu)轫樞蛄?。中的使用順序求和并行求和將流轉(zhuǎn)為并行流配置并行流線(xiàn)程池并行流內(nèi)部使用了默認(rèn)的，默認(rèn)的線(xiàn)程數(shù)量就是處理器的數(shù)量包括虛擬內(nèi)核通過(guò)得到。

【概念

并行流就是一個(gè)把內(nèi)容分成多個(gè)數(shù)據(jù)塊，并用不同的線(xiàn)程分別處理每一個(gè)數(shù)據(jù)塊的流。在java7之前，并行處理數(shù)據(jù)很麻煩，第一，需要明確的把包含數(shù)據(jù)的數(shù)據(jù)結(jié)構(gòu)分成若干子部分。第二，給每一個(gè)子部分分配一個(gè)獨(dú)立的線(xiàn)程。第三，適當(dāng)?shù)臅r(shí)候進(jìn)行同步，避免出現(xiàn)數(shù)據(jù)競(jìng)爭(zhēng)帶來(lái)的問(wèn)題，最后將每一個(gè)子部分的結(jié)果合并。在java7中引入了forkjoin框架來(lái)完成這些步驟，而java8中的stream接口可以讓你不費(fèi)吹灰之力就對(duì)數(shù)據(jù)執(zhí)行并行處理，而stream接口幕后正是使用的forkjoin框架。不過(guò)，對(duì)順序流調(diào)用parallel()并不意味著流本身有任何的變化。它在內(nèi)部實(shí)際上就是設(shè)了一個(gè)boolean標(biāo)志，表示你想讓parallel()之后的操作都并行執(zhí)行。類(lèi)似的你可以用sequential()將并行流變?yōu)轫樞蛄?。這兩個(gè)方法可以讓我們更細(xì)化的控制流。

eg.java8中stream的使用：

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

//并行求和
public static long parallelSum(long n){
    return Stream.iterate(1l,i -> i + 1)
            .limit(n)
            //將流轉(zhuǎn)為并行流
            .parallel()
            .reduce(0l,Long::sum);
}

【配置并行流線(xiàn)程池

并行流內(nèi)部使用了默認(rèn)的forkjoinPool，默認(rèn)的線(xiàn)程數(shù)量就是處理器的數(shù)量（包括虛擬內(nèi)核）,
通過(guò)：Runtime.getRuntime().availableProcessors() 得到。
通過(guò)：System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","12")來(lái)改變線(xiàn)程池大小。

【性能測(cè)試

我們不應(yīng)該理所當(dāng)然的任認(rèn)為多線(xiàn)程比順序執(zhí)行的效率更高，來(lái)看下面的例子：

public class Exercise {

    public static void main(String[] args) {
        long num = 1000_000_0;

        long st = System.currentTimeMillis();
        System.out.println("iterate順序" + sum(num) + ":" +(System.currentTimeMillis() - st));

        st = System.currentTimeMillis();
        System.out.println("iterate并行" + parallelSum(num) + ":" +(System.currentTimeMillis() - st));

        st = System.currentTimeMillis();
        System.out.println("迭代" + forSum(num) + ":" +(System.currentTimeMillis() - st));

        st = System.currentTimeMillis();
        System.out.println("LongStream并行" + longStreamParallelSum(num) + ":" +(System.currentTimeMillis() - st));

        st = System.currentTimeMillis();
        System.out.println("LongStream順序" + longStreamSum(num) + ":" +(System.currentTimeMillis() - st));
    }

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

    //并行求和
    public static long parallelSum(long n){
        return Stream.iterate(1l,i -> i + 1)
                .limit(n)
                //將流轉(zhuǎn)為并行流
                .parallel()
                .reduce(0l,Long::sum);
    }

    //迭代求和
    public static long forSum(long n){
        long result = 0;
        for(long i = 0 ;i <= n ; i++){
            result += i;
        }
        return result;
    }

    //longStream并行
    public static long longStreamParallelSum(long n){
        return LongStream.rangeClosed(1,n)
                .parallel()
                .reduce(0l,Long::sum);
    }

    //longStream順序執(zhí)行
    public static long longStreamSum(long n){
        return LongStream.rangeClosed(1,n)
                .reduce(0l,Long::sum);
    }
}

并行流執(zhí)行的時(shí)間比順序流和迭代執(zhí)行的要長(zhǎng)很多，兩個(gè)原因：

iterate()生成的是裝箱對(duì)象，必須要拆箱才能求和；

iterate()很難分成多個(gè)獨(dú)立的塊并行運(yùn)行，因?yàn)槊看螒?yīng)用這個(gè)函數(shù)都要依賴(lài)前一次的應(yīng)用的結(jié)果。數(shù)字列表在歸納的過(guò)程開(kāi)始時(shí)沒(méi)有準(zhǔn)備好，因而無(wú)法有效的把流劃分成小塊來(lái)并行處理。但是我們又標(biāo)記流為并行執(zhí)行，這就給順序執(zhí)行增加了開(kāi)銷(xiāo)，每一次的求和操作都新開(kāi)啟了一個(gè)線(xiàn)程。

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

LongStream.rangeClosed():

    1. 直接產(chǎn)生long類(lèi)型數(shù)據(jù)，沒(méi)有開(kāi)箱操作
    2. 生成數(shù)字范圍，容易拆分成獨(dú)立的小塊

由此可見(jiàn)，選擇適當(dāng)?shù)臄?shù)據(jù)結(jié)構(gòu)往往比并行化算法更重要。并行是有代價(jià)的。并行過(guò)程需要對(duì)流做遞歸劃分，把每個(gè)子流的操作分配到不同的線(xiàn)程，然后把這些操作的結(jié)果合并成一個(gè)值。但是多核之間移動(dòng)數(shù)據(jù)的代價(jià)比我們想象的要大，所以很重要的一點(diǎn)是保證再內(nèi)核中并行執(zhí)行的工作時(shí)間比內(nèi)核之間傳輸數(shù)據(jù)的時(shí)間要長(zhǎng)。

【正確的使用并行流

錯(cuò)誤使用并行流的首要原因就是使用的算法改變了共享變量的狀態(tài)，因?yàn)樾薷墓蚕碜兞恳馕吨?，而使用同步方法就?huì)使的并行毫無(wú)意義。以下是一些建議：

1. 測(cè)試，并行還是順序執(zhí)行最重要的基準(zhǔn)就是不停的測(cè)試性能。
2. 留意裝箱，自動(dòng)裝箱，拆箱會(huì)大大降低性能，java8提供了LongStream,IntStream,DoubleStream來(lái)避免這兩個(gè)操作。
3. 有些操作本身就是順序執(zhí)行要率高，例如：limit，findFirst等依賴(lài)元素順序的操作。
4. 當(dāng)執(zhí)行單個(gè)任務(wù)的成本高時(shí)使用并行，如果單個(gè)操作的成本很低，并行執(zhí)行反而會(huì)因?yàn)殚_(kāi)啟線(xiàn)程，標(biāo)記狀態(tài)等操作使得效率下降。
5. 小量數(shù)據(jù)不適用并行。
6. 考慮流中背后的數(shù)據(jù)結(jié)構(gòu)是否易于分解。ArrayList的拆分效率比LinkedList高得多，因?yàn)榍罢哂貌恢憷涂梢云骄鸱?。另外，range工廠(chǎng)方法的原始類(lèi)型數(shù)據(jù)流也可以快速分解。以下時(shí)流數(shù)據(jù)源的可分解性：
   - ArrayList:極佳
   - LinkedList:差
   - IntStream等：極佳
   - Stream.iterate:差
   - HashSet:好
   - TreeSet:好
7. 中間操作改變流的方法，涉及到排序就不適用并行。
8. 終端操作合并流的代價(jià)，涉及到排序就不適用并行。

【正確的使用并行

高并發(fā)、任務(wù)執(zhí)行時(shí)間短的業(yè)務(wù)，線(xiàn)程池線(xiàn)程數(shù)可以設(shè)置為CPU核數(shù)+1，減少線(xiàn)程上下文的切換

并發(fā)不高、任務(wù)執(zhí)行時(shí)間長(zhǎng)的業(yè)務(wù)要區(qū)分開(kāi)看：

假如是業(yè)務(wù)時(shí)間長(zhǎng)集中在IO操作上，也就是IO密集型的任務(wù)，因?yàn)镮O操作并不占用CPU，所以不要讓所有的CPU閑下來(lái)，可以加大線(xiàn)程池中的線(xiàn)程數(shù)目，讓CPU處理更多的業(yè)務(wù)

假如是業(yè)務(wù)時(shí)間長(zhǎng)集中在計(jì)算操作上，也就是計(jì)算密集型任務(wù)，這個(gè)就沒(méi)辦法了，和（1）一樣吧，線(xiàn)程池中的線(xiàn)程數(shù)設(shè)置得少一些，減少線(xiàn)程上下文的切換

并發(fā)高、業(yè)務(wù)執(zhí)行時(shí)間長(zhǎng)，解決這種類(lèi)型任務(wù)的關(guān)鍵不在于線(xiàn)程池而在于整體架構(gòu)的設(shè)計(jì)，看看這些業(yè)務(wù)里面某些數(shù)據(jù)是否能做緩存是第一步，增加服務(wù)器是第二步，至于線(xiàn)程池的設(shè)置，設(shè)置參考（2）。最后，業(yè)務(wù)執(zhí)行時(shí)間長(zhǎng)的問(wèn)題，也可能需要分析一下，看看能不能使用中間件對(duì)任務(wù)進(jìn)行拆分和解耦。