ForkJoinPool使用介紹

出處：RecursiveTask和RecursiveAction的使用 以及java 8 並行流和順序流

 

什么是Fork/Join框架
        Fork/Join框架是Java7提供了的一個用於並行執行任務的框架， 是一個把大任務分割成若干個小任務，最終匯總每個小任務結果后得到大任務結果的框架。

         我們再通過Fork和Join這兩個單詞來理解下Fork/Join框架，Fork就是把一個大任務切分為若干子任務並行的執行，Join就是合並這些子任務的執行結果，最后得到這個大任務的結果。比如計算1+2+。。＋10000，可以分割成10個子任務，每個子任務分別對1000個數進行求和，最終匯總這10個子任務的結果。Fork/Join的運行流程圖如下：

 

 

 

工作竊取算法
         工作竊取（work-stealing）算法是指某個線程從其他隊列里竊取任務來執行。工作竊取的運行流程圖如下：

        
        那么為什么需要使用工作竊取算法呢？假如我們需要做一個比較大的任務，我們可以把這個任務分割為若干互不依賴的子任務，為了減少線程間的競爭，於是把這些子任務分別放到不同的隊列里，並為每個隊列創建一個單獨的線程來執行隊列里的任務，線程和隊列一一對應，比如A線程負責處理A隊列里的任務。但是有的線程會先把自己隊列里的任務干完，而其他線程對應的隊列里還有任務等待處理。干完活的線程與其等着，不如去幫其他線程干活，於是它就去其他線程的隊列里竊取一個任務來執行。而在這時它們會訪問同一個隊列，所以為了減少竊取任務線程和被竊取任務線程之間的競爭，通常會使用雙端隊列，被竊取任務線程永遠從雙端隊列的頭部拿任務執行，而竊取任務的線程永遠從雙端隊列的尾部拿任務執行。
     

        工作竊取算法的優點是充分利用線程進行並行計算，並減少了線程間的競爭，其缺點是在某些情況下還是存在競爭，比如雙端隊列里只有一個任務時。並且消耗了更多的系統資源，比如創建多個線程和多個雙端隊列。

 

ForkJoinPool
        Java提供了ForkJoinPool來支持將一個任務拆分成多個“小任務”並行計算，再把多個“小任務”的結果合成總的計算結果。

        ForkJoinPool是ExecutorService的實現類，因此是一種特殊的線程池。ForkJoinPool提供了如下兩個常用的構造器。

　　 

// 創建一個包含parallelism個並行線程的ForkJoinPool
public ForkJoinPool(int parallelism)

//以Runtime.getRuntime().availableProcessors()的返回值作為parallelism來創建ForkJoinPool
public ForkJoinPool() ：

         創建ForkJoinPool實例后，可以釣魚ForkJoinPool的submit(ForkJoinTask<T> task)或者invoke(ForkJoinTask<T> task)來執行指定任務。其中ForkJoinTask代表一個可以並行、合並的任務。ForkJoinTask是一個抽象類，它有兩個抽象子類：RecursiveAction和RecursiveTask。

• RecursiveTask代表有返回值的任務
• RecursiveAction代表沒有返回值的任務。

RecursiveAction
下面以一個沒有返回值的大任務為例，介紹一下RecursiveAction的用法。

大任務是：打印0-100的數值。

小任務是：每次只能打印20個數值。

代碼執行

public class RaskDemo extends RecursiveAction {
    /**
     *  每個"小任務"最多只打印20個數
      */
    private static final int MAX = 20;

    private int start;
    private int end;

    public RaskDemo(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
        //當end-start的值小於MAX時，開始打印
        if((end-start) < MAX) {
            for(int i= start; i<end;i++) {
                System.out.println(Thread.currentThread().getName()+"i的值"+i);
            }
        }else {
            // 將大任務分解成兩個小任務
            int middle = (start + end) / 2;
            RaskDemo left = new RaskDemo(start, middle);
            RaskDemo right = new RaskDemo(middle, end);
            left.fork();
            right.fork();
        }
    }
}

public static void main(String[] args) throws Exception{
    // 創建包含Runtime.getRuntime().availableProcessors()返回值作為個數的並行線程的ForkJoinPool
    ForkJoinPool forkJoinPool = new ForkJoinPool();

    // 提交可分解的PrintTask任務
    forkJoinPool.submit(new RaskDemo(0, 1000));

    //阻塞當前線程直到 ForkJoinPool 中所有的任務都執行結束
    forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);

    // 關閉線程池
    forkJoinPool.shutdown();
}

運行結果

 

 

 　　從上面結果來看，ForkJoinPool啟動了四個線程來執行這個打印任務，我的計算機的CPU是四核的。大家還可以看到程序雖然打印了0-999這一千個數字，但是並不是連續打印的，這是因為程序將這個打印任務進行了分解，分解后的任務會並行執行，所以不會按順序打印。

 

RecursiveTask
下面以一個有返回值的大任務為例，介紹一下RecursiveTask的用法。

大任務是：計算隨機的1000個數字的和。

小任務是：每次只能70個數值的和。

public class RecursiveTaskDemo extends RecursiveTask<Integer> {

    /**
     *  每個"小任務"最多只打印70個數
     */
    private static final int MAX = 70;
    private int arr[];
    private int start;
    private int end;


    public RecursiveTaskDemo(int[] arr, int start, int end) {
        this.arr = arr;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        int sum = 0;
        // 當end-start的值小於MAX時候，開始打印
        if((end - start) < MAX) {
            for (int i = start; i < end; i++) {
                sum += arr[i];
            }
            return sum;
        }else {
            System.err.println("=====任務分解======");
            // 將大任務分解成兩個小任務
            int middle = (start + end) / 2;
            RecursiveTaskDemo left = new RecursiveTaskDemo(arr, start, middle);
            RecursiveTaskDemo right = new RecursiveTaskDemo(arr, middle, end);
            // 並行執行兩個小任務
            left.fork();
            right.fork();
            // 把兩個小任務累加的結果合並起來
            return left.join() + right.join();
        }
    }

}

@Test
    public void dfs() throws Exception{
        int arr[] = new int[1000];
        Random random = new Random();
        int total = 0;
        // 初始化100個數字元素
        for (int i = 0; i < arr.length; i++) {
            int temp = random.nextInt(100);
            // 對數組元素賦值,並將數組元素的值添加到total總和中
            total += (arr[i] = temp);
        }
        System.out.println("初始化時的總和=" + total);

        // 創建包含Runtime.getRuntime().availableProcessors()返回值作為個數的並行線程的ForkJoinPool
        ForkJoinPool forkJoinPool = new ForkJoinPool();

        // 提交可分解的PrintTask任務
//        Future<Integer> future = forkJoinPool.submit(new RecursiveTaskDemo(arr, 0, arr.length));
//        System.out.println("計算出來的總和="+future.get());


        Integer integer = forkJoinPool.invoke( new RecursiveTaskDemo(arr, 0, arr.length)  );
        System.out.println("計算出來的總和=" + integer);

        // 關閉線程池
        forkJoinPool.shutdown();
    }

 

 

從上面結果來看，ForkJoinPool將任務分解了15次，程序通過SumTask計算出來的結果，和初始化數組時統計出來的總和是相等的，這表明計算結果一切正常。

 

總結

第一步分割任務  

        首先我們需要有一個fork類來把大任務分割成子任務，有可能子任務還是很大，所以還需要不停的分割，直到分割出的子任務足夠小。

 

第二步執行任務並合並結果。

      分割的子任務分別放在雙端隊列里，然后幾個啟動線程分別從雙端隊列里獲取任務執行。子任務執行完的結果都統一放在一個隊列里，啟動一個線程從隊列里拿數據，然后合並這些數據。

 

能夠輕松的利用多個 CPU 提供的計算資源來協作完成一個復雜的計算任務，提高運行效率！

 

java8新的寫法

/**
     *並行流 與 順序流
     */
    @Test
    public void test03() {

        Instant start = Instant.now();
        LongStream.rangeClosed( 0,110 )
                //並行流
                .parallel()
                .reduce( 0,Long::sum );

        LongStream.rangeClosed( 0,110 )
                //順序流
                .sequential()
                .reduce( 0,Long::sum );

        Instant end = Instant.now();
        System.out.println("耗費時間"+ Duration.between( start,end ).toMillis());
}