1. 概述

在進行系統儲存專題的學習、總結、整理和寫作的過程中感覺有點累了，加上最近在做書稿的第二次校稿工作，系統儲存專題的學習和寫作進度明顯有些滯後，特別是編寫的Ceph MON知識點。所以後續1、2個月的時間打算換一個學習整理方向，之後再繼續恢復系統儲存專題的寫作。換一個什麼方向呢？嗯～～之前的執行緒專欄還有一部分坑沒有填上，就繼續填這部分的坑吧。

java.util.concurrent.ForkJoinPool由Java大師Doug Lea主持編寫，它可以將一個大的任務拆分成多個子任務進行並行處理，最後將子任務結果合併成最後的計算結果，並進行輸出。本文中對Fork/Join框架的講解，基於JDK1.8+中的Fork/Join框架實現，參考的Fork/Join框架主要原始碼也基於JDK1.8+。

這幾篇文章將試圖解釋Fork/Join框架的知識點，以便對自己、對各位讀者在併發程式的設計思路上進行一些啟發。文章將首先講解Fork/Join框架的基本使用，以及其中需要注意的使用要點；接著使用Fork/Join框架解決一些實際問題；最後再講解Fork/Join框架的工作原理。

2. Fork/Join框架基本使用

這裡是一個簡單的Fork/Join框架使用示例，在這個示例中我們計算了1-1001累加後的值：

/**
 * 這是一個簡單的Join/Fork計算過程，將1—1001數字相加
 */
public class TestForkJoinPool {

    private static final Integer MAX = 200
 
;

    static class MyForkJoinTask extends RecursiveTask<Integer> {
        // 子任務開始計算的值
        private Integer startValue;

        // 子任務結束計算的值
        private Integer endValue;

        public MyForkJoinTask(Integer startValue , Integer endValue) {
            this.startValue = startValue;
            this
 
.endValue = endValue;
        }

        @Override
        protected Integer compute() {
            // 如果條件成立，說明這個任務所需要計算的數值分為足夠小了
            // 可以正式進行累加計算了
            if(endValue - startValue < MAX) {
                System.out.println("開始計算的部分：startValue = " + startValue + ";endValue = " + endValue);
                Integer totalValue = 0;
                for(int index = this.startValue ; index <= this.endValue  ; index++) {
                    totalValue += index;
                }
                return totalValue;
            }
            // 否則再進行任務拆分，拆分成兩個任務
            else {
                MyForkJoinTask subTask1 = new MyForkJoinTask(startValue, (startValue + endValue) / 2);
                subTask1.fork();
                MyForkJoinTask subTask2 = new MyForkJoinTask((startValue + endValue) / 2 + 1 , endValue);
                subTask2.fork();
                return subTask1.join() + subTask2.join();
            }
        }
    }

    public static void main(String[] args) {
        // 這是Fork/Join框架的執行緒池
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Integer> taskFuture =  pool.submit(new MyForkJoinTask(1,1001));
        try {
            Integer result = taskFuture.get();
            System.out.println("result = " + result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace(System.out);
        }
    }
}

以上程式碼很簡單，在關鍵的位置有相關的註釋說明。這裡本文再對以上示例中的要點進行說明。首先看看以上示例程式碼的可能執行結果：

開始計算的部分：startValue = 1;endValue = 126
開始計算的部分：startValue = 127;endValue = 251
開始計算的部分：startValue = 252;endValue = 376
開始計算的部分：startValue = 377;endValue = 501
開始計算的部分：startValue = 502;endValue = 626
開始計算的部分：startValue = 627;endValue = 751
開始計算的部分：startValue = 752;endValue = 876
開始計算的部分：startValue = 877;endValue = 1001
result = 501501

2-1. 工作順序圖

下圖展示了以上程式碼的工作過程概要，但實際上Fork/Join框架的內部工作過程要比這張圖複雜得多，例如如何決定某一個recursive task是使用哪條執行緒進行執行；再例如如何決定當一個任務/子任務提交到Fork/Join框架內部後，是建立一個新的執行緒去執行還是讓它進行佇列等待。

所以如果不深入理解Fork/Join框架的執行原理，只是根據之上最簡單的使用例子觀察執行效果，那麼我們只能知道子任務在Fork/Join框架中被拆分得足夠小後，並且其內部使用多執行緒並行完成這些小任務的計算後再進行結果向上的合併動作，最終形成頂層結果。不急，一步一步來，我們先從這張概要的過程圖開始討論。

圖中最頂層的任務使用submit方式被提交到Fork/Join框架中，後者將前者放入到某個執行緒中執行，工作任務中的compute方法的程式碼開始對這個任務T1進行分析。如果當前任務需要累加的數字範圍過大（程式碼中設定的是大於200），則將這個計算任務拆分成兩個子任務（T1.1和T1.2），每個子任務各自負責計算一半的資料累加，請參見程式碼中的fork方法。如果當前子任務中需要累加的數字範圍足夠小（小於等於200），就進行累加然後返回到上層任務中。

2-2. ForkJoinPool建構函式

ForkJoinPool有四個建構函式，其中引數最全的那個建構函式如下所示：

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode)

• parallelism：可並行級別，Fork/Join框架將依據這個並行級別的設定，決定框架內並行執行的執行緒數量。並行的每一個任務都會有一個執行緒進行處理，但是千萬不要將這個屬性理解成Fork/Join框架中最多存在的執行緒數量，也不要將這個屬性和ThreadPoolExecutor執行緒池中的corePoolSize、maximumPoolSize屬性進行比較，因為ForkJoinPool的組織結構和工作方式與後者完全不一樣。而後續的討論中，讀者還可以發現Fork/Join框架中可存在的執行緒數量和這個引數值的關係並不是絕對的關聯（有依據但並不全由它決定）。

• factory：當Fork/Join框架建立一個新的執行緒時，同樣會用到執行緒建立工廠。只不過這個執行緒工廠不再需要實現ThreadFactory介面，而是需要實現ForkJoinWorkerThreadFactory介面。後者是一個函式式介面，只需要實現一個名叫newThread的方法。在Fork/Join框架中有一個預設的ForkJoinWorkerThreadFactory介面實現：DefaultForkJoinWorkerThreadFactory。

• handler：異常捕獲處理器。當執行的任務中出現異常，並從任務中被丟擲時，就會被handler捕獲。

• asyncMode：這個引數也非常重要，從字面意思來看是指的非同步模式，它並不是說Fork/Join框架是採用同步模式還是採用非同步模式工作。Fork/Join框架中為每一個獨立工作的執行緒準備了對應的待執行任務佇列，這個任務佇列是使用陣列進行組合的雙向佇列。即是說存在於佇列中的待執行任務，即可以使用先進先出的工作模式，也可以使用後進先出的工作模式。

  

  當asyncMode設定為ture的時候，佇列採用先進先出方式工作；反之則是採用後進先出的方式工作，該值預設為false：

  ......
  asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
  ......

ForkJoinPool還有另外兩個建構函式，一個建構函式只帶有parallelism引數，既是可以設定Fork/Join框架的最大並行任務數量；另一個建構函式則不帶有任何引數，對於最大並行任務數量也只是一個預設值——當前作業系統可以使用的CPU核心數量（Runtime.getRuntime().availableProcessors()）。實際上ForkJoinPool還有一個私有的、原生建構函式，之上提到的三個建構函式都是對這個私有的、原生建構函式的呼叫。

......
private ForkJoinPool(int parallelism,
                         ForkJoinWorkerThreadFactory factory,
                         UncaughtExceptionHandler handler,
                         int mode,
                         String workerNamePrefix) {
        this.workerNamePrefix = workerNamePrefix;
        this.factory = factory;
        this.ueh = handler;
        this.config = (parallelism & SMASK) | mode;
        long np = (long)(-parallelism); // offset ctl counts
        this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
    }
......

如果你對Fork/Join框架沒有特定的執行要求，可以直接使用不帶有任何引數的建構函式。也就是說推薦基於當前作業系統可以使用的CPU核心數作為Fork/Join框架內最大並行任務數量，這樣可以保證CPU在處理並行任務時，儘量少發生任務執行緒間的執行狀態切換（實際上單個CPU核心上的執行緒間狀態切換基本上無法避免，因為作業系統同時執行多個執行緒和多個程序）。

2-3. fork方法和join方法

Fork/Join框架中提供的fork方法和join方法，可以說是該框架中提供的最重要的兩個方法，它們和parallelism“可並行任務數量”配合工作，可以導致拆分的子任務T1.1、T1.2甚至TX在Fork/Join框架中不同的執行效果。例如TX子任務或等待其它已存在的執行緒執行關聯的子任務，或在執行TX的執行緒中“遞迴”執行其它任務，又或者啟動一個新的執行緒執行子任務……

fork方法用於將新建立的子任務放入當前執行緒的work queue佇列中，Fork/Join框架將根據當前正在併發執行ForkJoinTask任務的ForkJoinWorkerThread執行緒狀態，決定是讓這個任務在佇列中等待，還是建立一個新的ForkJoinWorkerThread執行緒執行它，又或者是喚起其它正在等待任務的ForkJoinWorkerThread執行緒執行它。

這裡面有幾個元素概念需要注意，ForkJoinTask任務是一種能在Fork/Join框架中執行的特定任務，也只有這種型別的任務可以在Fork/Join框架中被拆分執行和合並執行。ForkJoinWorkerThread執行緒是一種在Fork/Join框架中執行的特性執行緒，它除了具有普通執行緒的特性外，最主要的特點是每一個ForkJoinWorkerThread執行緒都具有一個獨立的任務等待佇列（work queue），這個任務佇列用於儲存在本執行緒中被拆分的若干子任務。

join方法用於讓當前執行緒阻塞，直到對應的子任務完成執行並返回執行結果。或者，如果這個子任務存在於當前執行緒的任務等待佇列（work queue）中，則取出這個子任務進行“遞迴”執行。其目的是儘快得到當前子任務的執行結果，然後繼續執行。

關注微信公眾號和今日頭條，精彩文章持續更新中。。。。。