本文如果您已經了解一般并行編程知識。了解 Java concurrent 部分如 ExecutorService 等相關內容。

雖說是 Java 的 ForkJoin 并行框架。但不要太在意 Java ，當中的思想在其他語言環境也是相同適用的。由于并發編程在本質上是一樣的。就好像怎樣找到優秀的 Ruby 程序猿？事實上要找的僅僅是一個優秀的程序猿。

當然，假設語言層面直接支持相關的語義會更好。

?

引言

Java? 語言從一開始就支持線程和并發性語義。

Java5 添加的并發工具又攻克了一般應用程序的并發需求。 Java6 、 Java7 又進一步補充了一些內容。原來的工具主要是粗粒度的并發。

比方每一個 web 請求由一個工作線程處理，在線程池分配任務。

而Java 7 中新引入的 ForkJoin 能夠處理更細粒度的并行計算。

?

早期的時候都是單核 cpu 環境。假設不是多核環境下。線程 / 進程并非 真正 的并行運行，主要用來表示異步運行效果。

單核 cpu 上，假如每一個任務全然是 cpu 密集的（沒有等待），那么這樣的偽并發并不會使計算變快。

僅僅有在真正的多核環境才干起到加速作用，而如今多核已經普及。甚至已經到了手機上！

介紹

ForkJoin 是適用于多核環境的輕量級并行框架。目標是在多核系統下，通過并行運算。充分利用多處理器。提高效率與加速執行。

ForkJoin 編程范式：將問題遞歸地分解為較小的子問題，并行處理這些子問題，然后合并結果，如：

if (my portion of the work is small enough) ??

? ? ?do the work directly?

else ??

? ? ?split my work into two pieces ??

? ? ?invoke the two pieces and wait for the results

?

ForkJoin 由一組工作線程組成。用來運行任務。核心是 work-stealing 算法 。能夠有大量任務，但實際僅僅有少量真正的物理線程。默認是機器的 cpu 數量，也可指定。非常多其他工作的算法都能夠在此基礎之上進行。

盡管起初直接使用它的人可能不多，但將來會被非常多框架在底層使用，由于是如此基礎，所以終于 ForkJoin 可能會無處不在。

一般而言，使用者僅僅須要關心兩個方法 fork()? 和 ?join() 。它們分別表示：子任務的異步運行和堵塞等待結果完畢。 ?

ForkJoin 框架的核心是 ForkJoinPool 類，實現了 work-stealing 算法，用于運行 ForkJoinTask 類型的任務（也就是依照該算法調度線程與任務，當然還負責解決好相關的一些其他問題）。

?

work-stealing 算法

work-stealing? 是一種任務調度方法，由多個工作線程組成。每一個工作線程用一個雙端隊列維護一組任務。

Fork 的時候是把任務加到隊列的頭部。而不像一般的線程池那樣是加到任務隊列末尾。工作線程選擇頭部最新的任務來運行。當工作線程沒有任務可運行時，它會嘗試從其他線程的任務隊列尾部竊取一個任務運行。假設沒有任務運行了而且竊取其他任務失敗，那么工作線程停止。

這樣的方法的長處是降低了爭用，由于工作線程從頭獲取任務，而竊取線程從尾部竊取任務。還有一個長處是遞歸的分治法使得早期產生的是較大的任務單元。而竊取到較大任務會進一步遞歸分解。因此也降低了尾部竊取的次數。

另外，父任務非常可能要等待子任務（ join ）。所以從隊列頭部子任務開始運行也是一種優化。

總之。它會使用有限的線程運行大量任務，同一時候保持各線程的任務都處于繁忙的運行狀態。而盡量不讓線程處于等待狀態。

為了做到這點可能會從其他線程的任務隊列中竊取任務來運行，所以叫 work-stealing 。

就像前面所說物理線程不能太多，過多的話切換管理開銷就會較大，還會消耗很多其它的內存等資源，而且沒有帶來不論什么優點。默認是用 cpu數量的線程數，普通情況都 比較合適（比方 Runtime.getRuntime().availableProcessors() 返回處理器的數量），但詳細的數值還和任務自身的特點有關，能夠通過不同參數測試比較一下。而任務能夠是大量的，由每一個線程的工作隊列維護。

ForkJoin 是簡化了一些開發人員的工作，假設不用 ForkJoin 。最原始的方式是自己手工切分任務并分別創建線程運行。

?

分治、并行、可伸縮的思考：

這三者 關系非常親熱。分治思想（ divide-and-conquer ）是一種簡單樸素的思想。非常多問題都能夠這樣解決。 ForkJoin 就相當于分治法的并行版本號。 ? 分治本身僅僅是解決這個問題的思想，既能夠順序運行也能夠并行運行，可是在并行環境中更加有效。由于能夠并行處理子問題。

而在并行方面，可并行處理問題要么是彼此全然獨立的問題，要么是可分解單獨處理的問題。可伸縮性又和是否能并行處理緊密相關。由于假設不能并行處理就要受到單機處理能力的限制，也就難以伸縮了。

ForkJoin 與 MapReduce 兩個并行計算框架的差別 ? ？

MapReduce 是把大數據集切分成小數據集，并行分布計算后再合并。

ForkJoin 是將一個問題遞歸分解成子問題。再將子問題并行運算后合并結果。

二者共同點：都是用于運行并行任務的。基本思想都是把問題分解為一個個子問題分別計算，再合并結果。應該說并行計算都是這樣的思想，彼此獨立的或可分解的。 從名字上看 Fork 和 Map 都有切分的意思。 Join 和 Reduce 都有合并的意思。比較類似。

差別：

1 ）環境差異。分布式 ?vs? 單機多核： ForkJoin 設計初衷針對單機多核（處理器數量非常多的情況）。 MapReduce 一開始就明白是針對非常多機器組成的集群環境的。也就是說一個是想充分利用多處理器，而還有一個是想充分利用非常多機器做分布式計算。這是兩種不同的的應用場景。有非常多差異，因此在細的編程模式方面有非常多不同。

2 ）編程差異： MapReduce 通常是：做較大粒度的切分，一開始就先切分好任務然后再運行，而且彼此間在最后合并之前不須要通信。這樣可伸縮性更好，適合解決巨大的問題，但限制也很多其它。 ForkJoin 能夠是較小粒度的切分，任務自己知道該怎樣切分自己，遞歸地切分到一組合適大小的子任務來運行。由于是一個 JVM 內，所以彼此間通信是非常easy的。更像是傳統編程方式。

ForkJoin 框架基本結構 ：

ForkJoinPool 本身實現了 ExecutorService 接口，負責調度運行 ForkJoinTask 。

ForkJoinTask 是提交給 ForkJoinPool? 運行的任務，本身也實現了 Future? 接口。

?

ForkJoinTask 有兩個子類 RecursiveAction 和 RecursiveTask 。

? RecursiveAction? 沒有返回值（僅僅需 fork ）； RecursiveTask 有返回值（須要合并）。

類似于 Runnable 和 ?Callable 一樣。沒有返回值一般意味著全部子任務都運行完了就可以，中間的子任務不須要 join 了。事實上要不要返回值都能夠實現。有返回值能夠直接合并。沒有返回值能夠把結果保存在共享的數據上。

?

而我們要做的是實現自己要完畢的任務，僅僅須要繼承其一，并覆蓋抽象方法 compute() 。在這種方法中實現自己的任務，遞歸分解任務。

?

ForkJoinPool 與一般的 ExecutorService 實現的區別 ？

ForkJoin 實現了 ExecutorService 接口，這個接口就是用來把任務交給線程池中的工作線程去運行。 ForkJoin 也是一個 ExecutorService ，但差別在于 ForkJoin 使用了 work-stealing 算法，見前面的介紹。

普通的線程池是按 FIFO 的方式運行，而 ForkJoin 優先運行（由其他任務）后創建子任務。

對于大部分 會 產生子任務的 任務 模式， ForkJoin 的處理 實現 會非常高效。假設設置了異步模式， ? ForkJoin 也可能適合運行事件類型（不須要 join ）的任務。

影響 ForkJoin 加速效果的因素

理想效果是核越多加速效果越好。

可是并行不一定更快，參數不正確還可能更慢：

1） ?? 并發數，即線程數。通常是可用的 cpu 數，默認就是這個，一般表現非常好。

2） ?? 任務切分的粒度。假設切分粒度等于總任務量，一個任務運行，就相當于單線程順序運行。每一個任務運行的計算量。太大的話加速效果有限。不能發揮到最好。相反，太小的話，消耗在任務管理的成本占了主要部分。導致還不如順序運行的快。 ?

須要適當平衡二者。由于還和任務本身的特定有關。所以能夠做個基準測試比較一下。

而總的運行時間還與任務的規模有關。

任務粒度應該適中，多大合適？好像在什么地方上看到說：經驗上單個任務為 100-10000 個基本指令，當然還和任務本身的特定有關。

?

個人感覺多核 cpu 僅僅適用于解決計算密集型應用，由于實際問題可能 IO 等其它方面的瓶頸，多核也還是無法充分利用的。

? ?

使用 ForkJoin 的步驟：

ForkJoin 框架替我們完畢了一些工作，那么我們使用時還要完畢哪些工作：

1） ?? 怎樣運行單個任務。假設僅僅切分出一個任務運行，就相當于單線程順序運行。

2） ?? 怎樣遞歸地切分任務（以及任務切分后是否須要合并結果）

3） ?? 切分粒度多少合適（最小任務單元）

這些詳細表如今：繼承 ForkJoinTask 的一個子類。并實現抽象方法 compute() 。

在這種方法中實現自己的任務，遞歸分解任務。

?

這些準備好之后就能夠啟動了：創建一個表示所有任務的 ForkJoinTask 對象，創建一個 ForkJoinPool 的實例。把 task 作為參數運行 ForkJoinPool 的 invoke 方法。

?

在 ForkJoin 任務外部運行總任務： execute 異步運行任務。沒有返回結果 void ； invoke 運行任務并等待返回結果。結果是特定類型。 submit 運行一個任務，返回 ForkJoinTask （實際上是作為 Future 對象返回）。一般應該在外部使用 invoke 調用運行總任務。

而 execute 和 submit 僅僅是為了實現 ExecutorService 規定的相關語義， invoke 是 ForkJoin 中特有的。

?

在 ForkJoinTask 內部遞歸運行的過程中： fork 是異步運行。 invoke 是等待任務運行完畢。

?

詳細實例：

多看看詳細演示樣例比較好。

1） ?? 合并排序演示樣例：

合并排序是常見的排序算法之中的一個。

演示樣例實現了對一個整數數組的合并排序。同一時候還演示了不同并發數（線程數）與不同數組大小的組合測試。

代碼在 <jdk_ home>/sample/ForkJoin/? 中。

?

2） ?? 把圖片模糊處理演示樣例：

一個圖片能夠被表示為一個 m*n 大小的整數數組，當中每一個整數表示一個像素（的顏色）。模糊處理之后的圖像還是一個相同大小的整數數組。處理過程是把原來的每一個像素與周圍像素的顏色求平均值就可以。

假設順序運行就是從頭到尾對每一個像素運行一次計算得到目標像素，由于每一個像素的計算是獨立的，所以能夠把這個整數數組切分成一塊一塊的子數組（即子任務）分別運行。任務不適合切分的過小。所以設定了一個常數閾值 10000 ，大小小于 10000 的子數組就直接運行，否則對半切分為兩個子數組的任務分別運行。 文章 ? 源碼

? ? ?? 在我自己的機器上 i3 處理器 ? （i3， 4cpu ），并行確實快了不少。

?

其他的比如：求最大值 max 、平均值 avg 、求和 sum 等聚合函數都是能夠分解計算的。

演示樣例中都是對數組的處理。比較常見的是對數組、集合進行并行地處理操作。但也不限于此。 ??

網上有一些 Fibonacci? 的演示樣例。但這些演示樣例并不適合展示 ForkJoin 。

?

Doug Lea 與 JSR-166

說到 Java 并發編程，就不能不說到 Doug Lea 與 JSR-166 。 Doug Lea 是并發編程方面的專家。紐約州立大學奧斯威戈分校的計算機教授。曾是 JCP 運行委員。是 JSR-166 的 leader 。

JSR-166 就是負責向 Java 語言中加入并發編程工具的，即我們見到的 java.util.concurrent 包（及子包）。還是《 ?Concurrent Programming in Java Design Principles and Patterns 》一書的作者，是這方面最早的書。

他還是知名的內存分配方法 dlmalloc 的作者，這是 C 語言中的動態內存分配函數 malloc 的一種普遍使用的實現。

? ?

參考資料：

• Java api? 及前面提到的兩個自帶演示樣例。
• A Java Fork/Join Framework ? （ pdf ）
• Java 理論與實踐: 應用 fork-join 框架
• Java 理論與實踐: 應用 fork-join 框架，第 2 部分
• JDK 7 中的 Fork/Join 模式
• Java Fork/Join for Parallel Programming
• Fork-Join Development in Java? SE
• Java Fork/Join + Groovy
• InfoQ 上 ForkJoin 相關文章

? 其他并行框架：

• Akka ?
• GPars ?