Fork / Join 框架簡介

Fork / Join 它可以將一個大的任務(wù)拆分成多個子任務(wù)進(jìn)行并行處理，最后將子任務(wù)結(jié)果合并成最后的計算結(jié)果，并進(jìn)行輸出。Fork / Join 框架要完成兩件事情。

Fork：把一個復(fù)雜任務(wù)進(jìn)行分拆，大事化小。

Join ：把分拆任務(wù)的結(jié)果進(jìn)行合并。

任務(wù)分割：首先 Fork/Join 框架需要把大的任務(wù)分割成足夠小的子任務(wù)，如果子任務(wù)比較大的話還要對子任務(wù)進(jìn)行繼續(xù)分割。

執(zhí)行任務(wù)并合并結(jié)果：分割的子任務(wù)分別放到雙端隊列里，然后幾個啟動線程分別從雙端隊列里獲取任務(wù)執(zhí)行。子任務(wù)執(zhí)行完的結(jié)果都放在另外一個隊列里，啟動一個線程從隊列里取數(shù)據(jù)，然后合并這些數(shù)據(jù)。

在 Java 的 Fork/Join 框架中，使用兩個類完成上述操作：

ForkJoinTask：我們要使用 Fork/Join 框架，首先需要創(chuàng)建一個 ForkJoin 任務(wù)。該類提供了在任務(wù)中執(zhí)行 fork 和 join 的機(jī)制。通常情況下我們不需要直接集成 ForkJoinTask 類，只需要繼承它的子類，F(xiàn)ork/Join 框架提供了兩個子類：

• RecursiveAction：用于沒有返回結(jié)果的任務(wù)；
• RecursiveTask:用于有返回結(jié)果的任務(wù)。

ForkJoinPool：ForkJoinTask 需要通過 ForkJoinPool 來執(zhí)行。

RecursiveTask：繼承后可以實現(xiàn)遞歸(自己調(diào)自己)調(diào)用的任務(wù)。

Fork/Join 框架的實現(xiàn)原理

ForkJoinPool 由 ForkJoinTask 數(shù)組和 ForkJoinWorkerThread 數(shù)組組成，F(xiàn)orkJoinTask 數(shù)組負(fù)責(zé)將存放以及將程序提交給 ForkJoinPool，而ForkJoinWorkerThread 負(fù)責(zé)執(zhí)行這些任務(wù)。

fork() 方法

fork() 方法的實現(xiàn)原理：當(dāng)我們調(diào)用 ForkJoinTask 的 fork 方法時，程序會把任務(wù)放在 ForkJoinWorkerThread 的 workQueue 中，異步地執(zhí)行這個任務(wù)，然后立即返回結(jié)果。源碼如下：

public final ForkJoinTask<V> fork() {
    Thread t;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
        ForkJoinPool.common.externalPush(this);
    return this;
}

push(Task) 方法把當(dāng)前任務(wù)存放在 ForkJoinTask 數(shù)組隊列里。然后再調(diào)用 ForkJoinPool 的 signalWork()方法喚醒或創(chuàng)建一個工作線程來執(zhí)行任務(wù)。源碼如下：

final void push(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a; ForkJoinPool p;
    int b = base, s = top, n;
    if ((a = array) != null) {    // ignore if queue removed
        int m = a.length - 1;     // fenced write for task visibility
        U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
        U.putOrderedInt(this, QTOP, s + 1);
        if ((n = s - b) <= 1) {
            if ((p = pool) != null)
                p.signalWork(p.workQueues, this);
        }
        else if (n >= m)
            growArray();
    }
}

join() 方法

join() 方法的主要作用是阻塞當(dāng)前線程并等待獲取結(jié)果。讓我們一起看看 ForkJoinTask 的 join() 方法的實現(xiàn)，源碼如下：

public final V join() {
    int s;
    if ((s = doJoin() & DONE_MASK) != NORMAL)
        reportException(s);
    return getRawResult();
}

它首先調(diào)用 doJoin 方法，通過 doJoin()方法得到當(dāng)前任務(wù)的狀態(tài)來判斷返回什么結(jié)果，任務(wù)狀態(tài)有 4 種：

已完成（NORMAL）、被取消（CANCELLED）、信號（SIGNAL）和出現(xiàn)異常（EXCEPTIONAL）

• 如果任務(wù)狀態(tài)是已完成，則直接返回任務(wù)結(jié)果。
• 如果任務(wù)狀態(tài)是被取消，則直接拋出 CancellationException。
• 如果 任務(wù)狀態(tài)是拋出異常，則直接拋出對應(yīng)的異常。

private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    //返回執(zhí)行狀態(tài)
    return (s = status) < 0 ? s :
    ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
       //取出任務(wù)
        (w = (wt = (ForkJoinWorkerThread)t).workQueue).
        //執(zhí)行任務(wù)
        tryUnpush(this) && (s = doExec()) < 0 ? s :
    wt.pool.awaitJoin(w, this, 0L) :
    externalAwaitDone();
}

final int doExec() {
    int s; boolean completed;
    if ((s = status) >= 0) {
        try {
            completed = exec();
        } catch (Throwable rex) {
            //記錄異常
            return setExceptionalCompletion(rex);
        }
        if (completed)
            //執(zhí)行順利，設(shè)置任務(wù)狀態(tài)為NORMAL
            s = setCompletion(NORMAL);
    }
    return s;
}

在 doJoin()方法流程如下:

1. 首先通過查看任務(wù)的狀態(tài)，看任務(wù)是否已經(jīng)執(zhí)行完成，如果執(zhí)行完成，則直接返回任務(wù)狀態(tài)。
2. 如果沒有執(zhí)行完，則從任務(wù)數(shù)組里取出任務(wù)并執(zhí)行。
3. 如果任務(wù)順利執(zhí)行完成，則設(shè)置任務(wù)狀態(tài)為 NORMAL，如果出現(xiàn)異常，則記錄異常，并將任務(wù)狀態(tài)設(shè)置為 EXCEPTIONAL。

Fork / Join 框架的異常處理

ForkJoinTask 在執(zhí)行的時候可能會拋出異常，但是我們沒辦法在主線程里直接捕獲異常，所以 ForkJoinTask 提供了 isCompletedAbnormally() 方法來檢查任務(wù)是否已經(jīng)拋出異?；蛞呀?jīng)被取消了，并且可以通過 ForkJoinTask 的 getException() 方法獲取異常。

getException 方法返回 Throwable 對象，如果任務(wù)被取消了則返回 CancellationException。如果任務(wù)沒有完成或者沒有拋出異常則返回 null。

入門案例

場景：生成一個計算任務(wù)，計算 1+2+3.........+1000，每 100 個數(shù)切分一個子任務(wù)

class MyTask extends RecursiveTask<Integer> {

    //拆分差值不能超過10，計算10以內(nèi)運算
    private static final Integer VALUE = 10;
    private int begin;//拆分起始值
    private int end;//拆分末尾值
    private int result;//結(jié)束返回值

    //創(chuàng)建有參構(gòu)造函數(shù)
    public MyTask(int begin, int end){
        this.begin = begin;
        this.end = end;
    }

    //拆分合并過程
    @Override
    protected Integer compute() {
        if (end-begin <= VALUE){
            for (int i = begin; i < end; i++) {
                result += i;
            }
        } else {
            //進(jìn)一步拆分
            int middle = (begin+end)/2;
            //拆分左邊
            MyTask task01 = new MyTask(begin,middle);
            //拆分右邊
            MyTask task02 = new MyTask(middle+1,end);
            //調(diào)用方法拆分
            task01.fork();
            task02.fork();
            //合并結(jié)果
            result = task01.join() + task02.join();
        }
        return result;
    }
}

public class ForkJoinDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //創(chuàng)建MyTask對象
        MyTask myTask = new MyTask(0,1000);
        //創(chuàng)建分支合并池對象
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Integer> forkJoinTask = forkJoinPool.submit(myTask);
        //獲取最終合并之后的結(jié)果
        try {
            Integer result = forkJoinTask.get();
            System.out.println(result);
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            //關(guān)閉池對象
            forkJoinPool.shutdown();
        }
    }
}

總結(jié)