一、ForkJoin介紹

     ForkJoin是Java7提供的原生多執行緒並行處理框架，其基本思想是將大人物分割成小任務，最後將小任務聚合起來得到結果。它非常類似於HADOOP提供的MapReduce框架，只是MapReduce的任務可以針對叢集內的所有計算節點，可以充分利用叢集的能力完成計算任務。ForkJoin更加類似於單機版的MapReduce。

二、工作竊取       

工作竊取(work-stealing)演算法是指某個執行緒從其他佇列裡竊取任務來執行。

 
        一個大任務分割為若干個互不依賴的子任務，為了減少執行緒間的競爭，把這些子任務分別放到不同的佇列裡，並未每個佇列建立一個單獨的執行緒來執行佇列裡的任務，執行緒和佇列一一對應。比如執行緒1負責處理1佇列裡的任務，2執行緒負責2佇列的。但是有的執行緒會先把自己佇列裡的任務幹完，而其他執行緒對應的佇列裡還有任務待處理。幹完活的執行緒與其等著，不如幫其他執行緒幹活，於是它就去其他執行緒的佇列裡竊取一個任務來執行。而在這時它們可能會訪問同一個佇列，所以為了減少竊取任務執行緒和被竊取任務執行緒之間的競爭，通常會使用雙端佇列，被竊取任務執行緒永遠從雙端佇列的頭部拿任務執行，而竊取任務執行緒永遠從雙端佇列的尾部拿任務執行。 

三、使用介紹

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

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

       第二步執行任務併合並結果。分割的子任務分別放在雙端佇列裡，然後幾個啟動執行緒分別從雙端佇列裡獲取任務執行。子任務執行完的結果都統一放在一個佇列裡，啟動一個執行緒從佇列裡拿資料，然後合併這些資料。

Fork/Join使用兩個類來完成以上兩件事情：

• ForkJoinTask：我們要使用ForkJoin框架，必須首先建立一個ForkJoin任務。它提供在任務中執行fork()和join()操作的機制，通常情況下我們不需要直接繼承ForkJoinTask類，而只需要繼承它的子類，Fork/Join框架提供了以下兩個子類：

• • RecursiveAction：用於沒有返回結果的任務。

  • RecursiveTask ：用於有返回結果的任務。

• ForkJoinPool ：ForkJoinTask需要通過ForkJoinPool來執行，任務分割出的子任務會新增到當前工作執行緒所維護的雙端佇列中，進入佇列的頭部。當一個工作執行緒的佇列裡暫時沒有任務時，它會隨機從其他工作執行緒的佇列的尾部獲取一個任務。

例項1使用recursiveTask

public class CountTask extends RecursiveTask<Integer> {
private static final long serialVersionUID = -3611254198265061729L;

public static final int threshold = 2;
private int start;
private int end;

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

@Override
protected Integer compute() {
    int sum = 0;

    //如果任務足夠小就計算任務
    boolean canCompute = (end - start) <= threshold;
    if (canCompute) {
        for (int i = start; i <= end; i++) {
            sum += i;
        }
    } else {
        // 如果任務大於閾值，就分裂成兩個子任務計算
        int middle = (start + end) / 2;
        CountTask leftTask = new CountTask(start, middle);
        CountTask rightTask = new CountTask(middle + 1, end);

        // 執行子任務
        leftTask.fork();
        rightTask.fork();

        //等待任務執行結束合併其結果
        int leftResult = leftTask.join();
        int rightResult = rightTask.join();

        //合併子任務
        sum = leftResult + rightResult;

    }

    return sum;
}

public static void main(String[] args) {
    ForkJoinPool forkjoinPool = new ForkJoinPool();

    //生成一個計算任務，計算1+2+3+4
    CountTask task = new CountTask(1, 100);

    //執行一個任務
    Future<Integer> result = forkjoinPool.submit(task);

    try {
        System.out.println(result.get());
    } catch (Exception e) {
        System.out.println(e);
    }
}

}
 

例項2使用recursiveAction

public class ForkJoinPoolTest {
/**
 * @param args
 * @throws Exception
 */
public static void main(String[] args) throws Exception {
    // 建立包含Runtime.getRuntime().availableProcessors()返回值作為個數的並行執行緒的ForkJoinPool
    ForkJoinPool forkJoinPool = new ForkJoinPool();
    // 提交可分解的PrintTask任務
    forkJoinPool.submit(new PrintTask(0, 200));
    forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);//阻塞當前執行緒直到 ForkJoinPool 中所有的任務都執行結束
    // 關閉執行緒池
    forkJoinPool.shutdown();
}

}

//RecursiveAction為ForkJoinTask的抽象子類，沒有返回值的任務
class PrintTask extends RecursiveAction {
// 每個"小任務"最多隻列印50個數
private static final int MAX = 50;

private int start;
private int end;

PrintTask(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() + " value i:"
                    + i);
        }
    } else {
        // 將大任務分解成兩個小任務
        int middle = (start + end) / 2;
        PrintTask left = new PrintTask(start, middle);
        PrintTask right = new PrintTask(middle, end);
        // 並行執行兩個小任務
        left.fork();
        right.fork();
    }
}
}