java-11-Stream優化並行流

 

並行流    多線程    把一個內容分成多個數據塊  不同線程分別處理每個數據塊的流

 

串行流   單線程  一個線程處理所有數據

 

java8 對並行流優化  StreamAPI 通過parallel()     並行流 

                                                        sequential()   順序流  

 

注意：

使用並行流並不是一定會提高效率，因為jvm對數據進行切片和切換線程也是需要時間的。

所以數據量越小，串行操作越快；數據量越大，並行操作效果越好。

 

 

StreamAPI 通過parallel()     並行流   底層是Fork/join 框架

 

Fork/join 框架   將任務分解成  若干小任務(分解到不可分解為止)    小任務的結果 join匯總

 

Fork/Join與傳統線程池的區別！

Fork/Join采用“工作竊取模式”，當執行新的任務時他可以將其拆分成更小的任務執行，並將小任務加到線程隊列中，然后再從一個隨即線程中偷一個並把它加入自己的隊列中。

就比如兩個CPU上有不同的任務，這時候A已經執行完，B還有任務等待執行，這時候A就會將B隊尾的任務偷過來，加入自己的隊列中，對於傳統的線程，ForkJoin更有效的利用的CPU資源！

 

 

ForkJoin的實現：實現這個框架需要繼承RecursiveTask 或者 RecursiveAction ，RecursiveTask是有返回值的，相反Action則沒有

 

溫馨提示: 電腦性能不好 不要run  很消耗電腦壽命

package com.wf.zhang.java8.stream;

import java.util.concurrent.RecursiveTask;


public class ForkJoinCalculate extends RecursiveTask<Long> {

    private static final long serialVersionUID = 13475679780L;
    
    private long start;
    private long end;
    
    private static final long THRESHOLD = 10000L; //臨界值

    //計算從start-end之和
    public ForkJoinCalculate(long start, long end) {
        this.start = start;
        this.end = end;
    }
    
    @Override
    protected Long compute() {
        long length = end - start;
        
        if(length <= THRESHOLD){
            long sum = 0;
            
            for (long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        }else{
            long middle = (start + end) / 2;
            
            ForkJoinCalculate left = new ForkJoinCalculate(start, middle);
            left.fork(); //拆分，並將該子任務壓入線程隊列
            
            ForkJoinCalculate right = new ForkJoinCalculate(middle+1, end);
            right.fork();
            //匯總
            return left.join() + right.join();
        }
        
    }

}

package com.wf.zhang.java8.stream;

import org.junit.Test;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;

public class TestForkJoin {

    @Test
    public void test1(){
        long start = System.currentTimeMillis();

        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkJoinCalculate(0L, 10000000000L);

        long sum = pool.invoke(task);
        System.out.println(sum);

        long end = System.currentTimeMillis();

        System.out.println("耗費的時間為: " + (end - start));
    }

    //java8 優化並行流
    @Test
    public void test3(){
        long start = System.currentTimeMillis();

        Long sum = LongStream.rangeClosed(0L, 10000000000L)
                             .parallel()
                             .sum();

        System.out.println(sum);

        long end = System.currentTimeMillis();

        System.out.println("耗費的時間為: " + (end - start));
    }
}

 

使用java8優化並行流         0L, 10000000000L 相加   

　　　　　　LongStream.rangeClosed ()    生成連續的Long類型的數   

　　　　　　　　　　　　　　 parellel()    並行流   底層是Fork/join 框架