前言

感謝王寶令老師極客時間的 課程，通俗易懂，這里再次推薦

哎，這篇文章敲了一遍沒看懂……

背景

前幾篇文章我們介紹了線程池，F(xiàn)uture 、CompletableFuture 和CompletionService （其中后兩者待補充）。仔細觀察你會發(fā)現(xiàn)這些工具類都是在幫我們站在任務(wù)的視角來解決并發(fā)問題，而不是讓我們糾纏在線程之間的如何協(xié)作細節(jié)上（比如線程之間如何等待、通知等），對于簡單的并行任務(wù)，可以通過線程池+Future 的方案來解決，如果任務(wù)之間有聚合關(guān)系，無論是AND聚合還是OR 聚合，都可以通過CompletableFuture 來解決，而批量的并行任務(wù)，則可以通過CompletionService 來解決。

我們一直講，并發(fā)編程可以分為三個層面的問題，分別是分工、協(xié)作、和互斥，當(dāng)你關(guān)注于任務(wù)的時候，你會發(fā)現(xiàn)你的視角已經(jīng)從并發(fā)編程的細節(jié)中跳出來了，你應(yīng)用的更多的是現(xiàn)實世界的思維模式，類比的往往是現(xiàn)實世界里的分工，所以我把線程池、Future 、CompletableFuture 和CompletionService 都列到了分工里面。

下面我們用現(xiàn)實世界里的工作流程圖描述了并發(fā)編程領(lǐng)域里的簡單并行任務(wù)、聚合任務(wù)和批量執(zhí)行任務(wù)，輔以流程圖，相信你一定能將你的思維模式轉(zhuǎn)換到顯示世界里來。

img

從上到下，依次為簡單并行任務(wù)、聚合任務(wù)和批量并行任務(wù)示意圖

上面提到的簡單并行、聚合、批量執(zhí)行這三種任務(wù)模型，基本上能夠覆蓋日常工作中的并發(fā)場景了，但是還是不夠全面，因為還有一種“分治”的任務(wù)模型沒有覆蓋到?！胺种巍?，顧名思義，分而治之，是一種解決復(fù)雜問題的思維方法和模式，具體來講，指的是把一個復(fù)雜的問題分解成多個相似的子問題，然后再把子問題分解成更小的子問題，直到子問題簡單到可以直接求解。理論上來講，解決每一個問題都對應(yīng)著一個任務(wù)，所以對于問題的分治，實際上就是對于任務(wù)的分治。

分治思想在很多領(lǐng)域都有廣泛應(yīng)用，例如算法領(lǐng)域有分治算法，（歸并排序，快速排序都屬于分治算法，二分法查找也屬于一種分治算法）；大數(shù)據(jù)領(lǐng)域的知名框架，MapReduce 背后的思想也是分治。既然分治這種任務(wù)模型如此普遍，那Java 顯然也需要支持，Java 并發(fā)包里提供了一種叫做Fork/Join 的 并行框架，就是用來支持分治這種任務(wù)模型的。

分治任務(wù)模型

這里你需要深入了解下分治任務(wù)模型，分治任務(wù)模型可以分為兩個階段 一個階段是任務(wù)分解，也就是將任務(wù)迭代的分解為子任務(wù)，直至子任務(wù)可以直接計算出結(jié)果。另一個階段是結(jié)果合并，即逐層合并子任務(wù)的執(zhí)行結(jié)果，直至獲取最終結(jié)果。下圖是一個簡化版的分治任務(wù)模型：

img

在這個分治任務(wù)模型里，任務(wù)和分解后的子任務(wù)具有相似性，這種相似性往往體現(xiàn)在任務(wù)和子任務(wù)的算法是相同的，但是計算的數(shù)據(jù)規(guī)模是不同的，具備這種相似性的問題，我們往往都采用遞歸的算法。

Fork/Join 的使用

Fork/Join 是一個并行計算框架，主要用來支持分治模型的，這個計算框架里的Fork 對應(yīng)的是分治任務(wù)模型里的分解，Join 對應(yīng)的結(jié)果合并。Fork/Join 計算框架主要包括兩部分，一部分是分治任務(wù)的線程池ForkJoinPool ，另一部分是分治任務(wù)ForkJoinTask 。這兩部分的關(guān)系類似于ThreadPoolExecutor 和Runnable 的關(guān)系，都可以理解為提交任務(wù)到線程池，只不過分治任務(wù)有自己獨特的類型ForkJoinTask 。

ForkJoinTask 是一個抽象類，他的方法很多，最核心的是fork() 方法和 join() 方法 ，其中fork() 方法會異步的執(zhí)行一個子任務(wù)，而join() 方法則會阻塞當(dāng)前線程來等待子任務(wù)的執(zhí)行結(jié)果。ForkJoinTask 有兩個子類RecursiveAction 和RecursiveTask ，通過名字你就應(yīng)該知道，他們都是遞歸的方式來處理分治任務(wù)的。這兩個子類都定義了抽象的方法compute()， 不過區(qū)別是RecursiveAction 定義的 compute() 沒有返回值，而 RecursiveTask 定義的 compute() 方法是有返回值的。 這兩個子類也是抽象類，在使用的時候需要你定義子類去擴展。

ForkJoinPool 工作原理

Fork/Join 并行計算的核心組件是ForkJoinPool ，所以下面我們來簡單介紹一下ForkJoinPool 的工作原理。

通過前面文章的學(xué)習(xí)，你應(yīng)該已經(jīng)知道了ThreadPoolExecutor 本質(zhì)上是一個生產(chǎn)者-消費者模式的實現(xiàn)，內(nèi)部有一個任務(wù)隊列，這個隊列是消費者生產(chǎn)者通信的媒介，ThreadPoolExecutor 可以有多個工作線程，但是這些工作線程都共享一個工作隊列。

ForkJoinPool 本質(zhì)上也是一個消費者-生產(chǎn)者模型的實現(xiàn)，但是更加智能，你可以參考下面的ForkJoinPool 工作原理來理解，F(xiàn)orkJoinPool 內(nèi)部只有一個任務(wù)隊列，而ForkJoinPool 內(nèi)部有多個任務(wù)隊列，當(dāng)我們通過ForkJoinPool 的invoke() 或者submit() 方法提交任務(wù)的時候，F(xiàn)orkJoinPool 根據(jù)一定的路由規(guī)則把任務(wù)提交到一個任務(wù)隊列中，如果任務(wù)再執(zhí)行過程中會創(chuàng)建出子任務(wù)，那么子任務(wù)會提交到工作線程對應(yīng)的任務(wù)隊列中。

如果工作線程對應(yīng)的任務(wù)隊列空了，是不是就沒活干了呢？不是的，F(xiàn)orkJoinPool 支持一種叫做“任務(wù)竊取”的機制， 如果工作線程空閑了，那么他可以竊取” 其他工作任務(wù)隊列中的任務(wù)，例如下圖中，線程 T2 對應(yīng)的任務(wù)隊列已經(jīng)空了，他可以竊取”線程 T1 對 應(yīng)的任務(wù)隊列中的任務(wù)，如此一來所有的工作線程都閑不下來了。

ForkJoinPool 中的任務(wù)隊列采用的是雙端隊列，工作線程正常獲取任務(wù)和竊取任務(wù)”分別是從任務(wù)隊列不同的端消費， 這樣能避免不必要的數(shù)據(jù)競爭，我們這里僅僅是簡化后的原理，F(xiàn)orkJoinPool 的實現(xiàn)遠比我們這里介紹的復(fù)雜，如果你感興趣，建議去看它的源碼。

img

模擬 MapReduce 統(tǒng)計單詞數(shù)量

學(xué)習(xí)MapReduce 有一個入門程序，統(tǒng)計一個文件里面每個單詞的數(shù)量，下面我們來看看如何用Fork/Join 并行計算框架實現(xiàn)。

這里省略實現(xiàn)……………………

總結(jié)

Fork/Join 并行計算框架主要解決的是分治任務(wù)。分治的核心思想是“分而治之”：將一個大的任務(wù)拆分成小任務(wù)去解決，然后再把子任務(wù)的結(jié)果聚合起來從而得到最終的結(jié)果。這個過程處理非常類似于大數(shù)據(jù)處理中的MapReduce ，所以你可以把Fork/Join 看作單機版的MapReduce 。

Fork/Join 并行計算框架的核心組件是ForkJoinPool 。ForkJoinPool 支持任務(wù)竊取機制，能夠讓所以的線程工作量基本均衡，不會出現(xiàn)有的線程很忙，有的線程很閑，所以性能很好。Java 1.8 提供的 Stream API 里面并行流也是以ForkJoinPool 為基礎(chǔ)的，不過需要注意的是，默認情況下所有的并行流計算都共享一個ForkJoinPool ，這個共享的ForkJoinPool 默認的線程數(shù)是CPU 的核數(shù)，如果所有的并行流計算都是CPU 密集型計算的話，完全沒有問題，但是如果存在I/O 密集型的并行流計算，那么很可能會因為一個慢的I/O 計算而拖慢整個系統(tǒng)的性能，所以建議用不同的ForkJoinPool 執(zhí)行不同類型的計算任務(wù)。