ForkJoin框架原理

ForkJoin框架是從jdk1.7中引入的新特性,它同ThreadPoolExecutor一樣，也實現(xiàn)了Executor和ExecutorService接口。它使用了一個無限隊列來保存需要執(zhí)行的任務，而線程的數(shù)量則是通過構造函數(shù)傳入，如果沒有向構造函數(shù)中傳入指定的線程數(shù)量，那么當前計算機可用的CPU數(shù)量會被設置為線程數(shù)量作為默認值。

ForkJoinPool主要使用?分治法(Divide-and-Conquer Algorithm)?來解決問題。典型的應用比如快速排序算法。這里的要點在于，F(xiàn)orkJoinPool能夠使用相對較少的線程來處理大量的任務。比如要對1000萬個數(shù)據(jù)進行排序，那么會將這個任務分割成兩個500萬的排序任務和一個針對這兩組500萬數(shù)據(jù)的合并任務。以此類推，對于500萬的數(shù)據(jù)也會做出同樣的分割處理，到最后會設置一個閾值來規(guī)定當數(shù)據(jù)規(guī)模到多少時，停止這樣的分割處理。比如，當元素的數(shù)量小于10時，會停止分割，轉而使用插入排序對它們進行排序。那么到最后，所有的任務加起來會有大概200萬+個。問題的關鍵在于，對于一個任務而言，只有當它所有的子任務完成之后，它才能夠被執(zhí)行。

所以當使用ThreadPoolExecutor時，使用分治法會存在問題，因為ThreadPoolExecutor中的線程無法向任務隊列中再添加一個任務并在等待該任務完成之后再繼續(xù)執(zhí)行。而使用ForkJoinPool就能夠解決這個問題，它就能夠讓其中的線程創(chuàng)建新的任務，并掛起當前的任務，此時線程就能夠從隊列中選擇子任務執(zhí)行。

那么使用ThreadPoolExecutor或者ForkJoinPool，性能上會有什么差異呢？

首先，使用ForkJoinPool能夠使用數(shù)量有限的線程來完成非常多的具有父子關系的任務，比如使用4個線程來完成超過200萬個任務。但是，使用ThreadPoolExecutor時，是不可能完成的，因為ThreadPoolExecutor中的Thread無法選擇優(yōu)先執(zhí)行子任務，需要完成200萬個具有父子關系的任務時，也需要200萬個線程，很顯然這是不可行的，也是很不合理的?。?/p>

工作竊取算法

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

工作竊取算法的優(yōu)點：充分利用線程進行并行計算，并減少了線程間的競爭。

工作竊取算法的缺點：在某些情況下還是存在競爭，比如雙端隊列里只有一個任務時。并且該算法會消耗更多的系統(tǒng)資源，比如創(chuàng)建多個線程和多個雙端隊列。

Fork/Join框架局限性：

對于Fork/Join框架而言，當一個任務正在等待它使用Join操作創(chuàng)建的子任務結束時，執(zhí)行這個任務的工作線程查找其他未被執(zhí)行的任務，并開始執(zhí)行這些未被執(zhí)行的任務，通過這種方式，線程充分利用它們的運行時間來提高應用程序的性能。為了實現(xiàn)這個目標，F(xiàn)ork/Join框架執(zhí)行的任務有一些局限性。

（1）任務只能使用Fork和Join操作來進行同步機制，如果使用了其他同步機制，則在同步操作時，工作線程就不能執(zhí)行其他任務了。比如，在Fork/Join框架中，使任務進行了睡眠，那么，在睡眠期間內，正在執(zhí)行這個任務的工作線程將不會執(zhí)行其他任務了。（2）在Fork/Join框架中，所拆分的任務不應該去執(zhí)行IO操作，比如：讀寫數(shù)據(jù)文件。（3）任務不能拋出檢查異常，必須通過必要的代碼來出來這些異常。

ForkJoin框架的實現(xiàn)

ForkJoin框架中一些重要的類如下所示。

ForkJoinPool 框架中涉及的主要類如下所示。

1.ForkJoinPool類

實現(xiàn)了ForkJoin框架中的線程池，由類圖可以看出，F(xiàn)orkJoinPool類實現(xiàn)了線程池的Executor接口。

我們也可以從下圖中看出ForkJoinPool的類圖關系。

其中，可以使用Executors.newWorkStealPool()方法創(chuàng)建ForkJoinPool。

ForkJoinPool中提供了如下提交任務的方法。

public?void?execute(ForkJoinTask?task)
public?void?execute(Runnable?task)
public??T?invoke(ForkJoinTask?task)
public??List>?invokeAll(Collection>?tasks)?
public??ForkJoinTask?submit(ForkJoinTask?task)
public??ForkJoinTask?submit(Callable?task)
public??ForkJoinTask?submit(Runnable?task,?T?result)
public?ForkJoinTask?submit(Runnable?task)

2.ForkJoinWorkerThread類

實現(xiàn)ForkJoin框架中的線程。

3.ForkJoinTask 類

ForkJoinTask封裝了數(shù)據(jù)及其相應的計算，并且支持細粒度的數(shù)據(jù)并行。ForkJoinTask比線程要輕量，F(xiàn)orkJoinPool中少量工作線程能夠運行大量的ForkJoinTask。

ForkJoinTask類中主要包括兩個方法fork()和join()，分別實現(xiàn)任務的分拆與合并。

fork()方法類似于Thread.start()，但是它并不立即執(zhí)行任務，而是將任務放入工作隊列中。跟Thread.join()方法不同，F(xiàn)orkJoinTask的join()方法并不簡單的阻塞線程，而是利用工作線程運行其他任務，當一個工作線程中調用join()，它將處理其他任務，直到注意到目標子任務已經完成。

我們可以使用下圖來表示這個過程。

ForkJoinTask有3個子類：

• RecursiveAction：無返回值的任務。

• RecursiveTask：有返回值的任務。

• CountedCompleter：完成任務后將觸發(fā)其他任務。

4.RecursiveTask 類

有返回結果的ForkJoinTask實現(xiàn)Callable。

5.RecursiveAction類

無返回結果的ForkJoinTask實現(xiàn)Runnable。

6.CountedCompleter 類

在任務完成執(zhí)行后會觸發(fā)執(zhí)行一個自定義的鉤子函數(shù)。

ForkJoin示例程序

package?io.binghe.concurrency.example.aqs;
?
import?lombok.extern.slf4j.Slf4j;
import?java.util.concurrent.ForkJoinPool;
import?java.util.concurrent.Future;
import?java.util.concurrent.RecursiveTask;
@Slf4j
public?class?ForkJoinTaskExample?extends?RecursiveTask<Integer>?{
????public?static?final?int?threshold?=?2;
????private?int?start;
????private?int?end;
????public?ForkJoinTaskExample(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;
????????????ForkJoinTaskExample?leftTask?=?new?ForkJoinTaskExample(start,?middle);
????????????ForkJoinTaskExample?rightTask?=?new?ForkJoinTaskExample(middle?+?1,?end);
?
????????????//?執(zhí)行子任務
????????????leftTask.fork();
????????????rightTask.fork();
?
????????????//?等待任務執(zhí)行結束合并其結果
????????????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
????????ForkJoinTaskExample?task?=?new?ForkJoinTaskExample(1,?100);
?
????????//執(zhí)行一個任務
????????Future?result?=?forkjoinPool.submit(task);
?
????????try?{
????????????log.info("result:{}",?result.get());
????????}?catch?(Exception?e)?{
????????????log.error("exception",?e);
????????}
????}
}

特別推薦一個分享架構+算法的優(yōu)質內容，還沒關注的小伙伴，可以長按關注一下：

  
長按訂閱更多精彩▼
如有收獲，點個在看，誠摯感謝