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

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

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

【概念

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

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);
}

【配置并行流線程池

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

【性能測試

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

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í)行的時間比順序流和迭代執(zhí)行的要長很多，兩個原因：

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

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

【使用更有針對性的的方法

LongStream.rangeClosed():

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

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

【正確的使用并行流

錯誤使用并行流的首要原因就是使用的算法改變了共享變量的狀態(tài)，因為修改共享變量意味著同步，而使用同步方法就會使的并行毫無意義。以下是一些建議：

1. 測試，并行還是順序執(zhí)行最重要的基準就是不停的測試性能。
2. 留意裝箱，自動裝箱，拆箱會大大降低性能，java8提供了LongStream,IntStream,DoubleStream來避免這兩個操作。
3. 有些操作本身就是順序執(zhí)行要率高，例如：limit，findFirst等依賴元素順序的操作。
4. 當執(zhí)行單個任務(wù)的成本高時使用并行，如果單個操作的成本很低，并行執(zhí)行反而會因為開啟線程，標記狀態(tài)等操作使得效率下降。
5. 小量數(shù)據(jù)不適用并行。
6. 考慮流中背后的數(shù)據(jù)結(jié)構(gòu)是否易于分解。ArrayList的拆分效率比LinkedList高得多，因為前者用不著便利就可以平均拆分。另外，range工廠方法的原始類型數(shù)據(jù)流也可以快速分解。以下時流數(shù)據(jù)源的可分解性：
   - ArrayList:極佳
   - LinkedList:差
   - IntStream等：極佳
   - Stream.iterate:差
   - HashSet:好
   - TreeSet:好
7. 中間操作改變流的方法，涉及到排序就不適用并行。
8. 終端操作合并流的代價，涉及到排序就不適用并行。

【正確的使用并行

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

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

假如是業(yè)務(wù)時間長集中在IO操作上，也就是IO密集型的任務(wù)，因為IO操作并不占用CPU，所以不要讓所有的CPU閑下來，可以加大線程池中的線程數(shù)目，讓CPU處理更多的業(yè)務(wù)

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

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