1. Fork/Join是什麼

  Oracle的官方給出的定義是：Fork/Join框架是一個實現了ExecutorService介面的多執行緒處理器。它可以把一個大的任務劃分為若干個小的任務併發執行，充分利用可用的資源，進而提高應用的執行效率。

  Fork/Join實現了ExecutorService，所以它的任務也需要放線上程池中執行。它的不同在於它使用了工作竊取演算法，空閒的執行緒可以從滿負荷的執行緒中竊取任務來幫忙執行。（我個人理解的工作竊取大意就是：由於執行緒池中的每個執行緒都有一個佇列，而且執行緒間互不影響。那麼執行緒每次都從自己的任務佇列的頭部獲取一個任務出來執行。如果某個時候一個執行緒的任務佇列空了，而其餘的執行緒任務佇列中還有任務，那麼這個執行緒就會從其他執行緒的任務佇列中取一個任務出來幫忙執行。就像偷取了其他人的工作一樣）

Fork/Join框架的核心是繼承了AbstractExecutorService的ForkJoinPool類，它保證了工作竊取演算法和ForkJoinTask的正常工作。

下面是引用Oracle官方定義的原文：

The fork/join framework is an implementation of the ExecutorService interface that helps you take advantage of multiple processors. It is designed for work that can be broken into smaller pieces recursively. The goal is to use all the available processing power to enhance the performance of your application.

As with any ExecutorService implementation, the fork/join framework distributes tasks to worker threads in a thread pool. The fork/join framework is distinct because it uses a work-stealing algorithm. Worker threads that run out of things to do can steal tasks from other threads that are still busy.

The center of the fork/join framework is the ForkJoinPool class, an extension of the AbstractExecutorService class. ForkJoinPool implements the core work-stealing algorithm and can execute ForkJoinTask processes.

2. Fork/Join的基本用法

（1）Fork/Join基類

  上文已經提到，Fork/Join就是要講一個大的任務分割成若干小的任務，所以第一步當然是要做任務的分割，大致方式如下：

1. if (這個任務足夠小){ 
2.   執行要做的任務 
3. } else { 
4.   將任務分割成兩小部分 
5.   執行兩小部分並等待執行結果 
6. } 

要實現FrokJoinTask我們需要一個繼承了RecursiveTask或RecursiveAction的基類，並根據自身業務情況將上面的程式碼放入基類的coupute方法中。RecursiveTask和RecursiveAction都繼承了FrokJoinTask，它倆的區別就是RecursiveTask有返回值而RecursiveAction沒有。下面是我做的一個選出字串列表中還有"a"的元素的Demo：

1. @Override
2. protected List<String> compute() { 
3.     // 當end與start之間的差小於閾值時，開始進行實際篩選
4.     if (end - this.start < threshold) { 
5.         List<String> temp = list.subList(this.start, end); 
6.         return temp.parallelStream().filter(s -> s.contains("a")).collect(Collectors.toList()); 
7.     } else { 
8.         // 如果當end與start之間的差大於閾值時
9.         // 將大任務分解成兩個小任務。
10.         int middle = (this.start + end) / 2; 
11.         ForkJoinTest left = new ForkJoinTest(list, this.start, middle, threshold); 
12.         ForkJoinTest right = new ForkJoinTest(list, middle, end, threshold); 
13.         // 並行執行兩個“小任務”
14.         left.fork(); 
15.         right.fork(); 
16.         // 把兩個“小任務”的結果合併起來
17.         List<String> join = left.join(); 
18.         join.addAll(right.join()); 
19.         return join; 
20.     } 
21. } 

（2）執行類

做好了基類就可以開始呼叫了，呼叫時首先我們需要Fork/Join執行緒池ForkJoinPool，然後向執行緒池中提交一個ForkJoinTask並得到結果。ForkJoinPool的submit方法的入參是一個ForkJoinTask，返回值也是一個ForkJoinTask，它提供一個get方法可以獲取到執行結果。

程式碼如下：

1. ForkJoinPool pool = new ForkJoinPool(); 
2. // 提交可分解的ForkJoinTask任務
3. ForkJoinTask<List<String>> future = pool.submit(forkJoinService); 
4. System.out.println(future.get()); 
5. // 關閉執行緒池
6. pool.shutdown(); 

就這樣我們就完成了一個簡單的Fork/Join的開發。

提示：Java8中java.util.Arrays的parallelSort()方法和java.util.streams包中封裝的方法也都用到了Fork/Join。（細心的讀者可能注意到我在Fork/Join中也有用到stream，所以其實這個Fork/Join是多餘的，因為stream已經實現了Fork/Join，不過這只是一個Demo展示，沒有任何實際用處也就無所謂了）

引用官方原文：

One such implementation, introduced in Java SE 8, is used by the java.util.Arrays class for its parallelSort() methods. These methods are similar to sort(), but leverage concurrency via the fork/join framework. Parallel sorting of large arrays is faster than sequential sorting when run on multiprocessor systems.

Another implementation of the fork/join framework is used by methods in the java.util.streams package, which is part of Project Lambda scheduled for the Java SE 8 release.

附完整程式碼以便以後參考：

1. 定義抽象類（用於拓展，此例中沒有實際作用，可以不定義此類）：

1. import java.util.concurrent.RecursiveTask; 
2. /** 
3.  * Description: ForkJoin介面 
4.  * Designer: jack 
5.  * Date: 2017/8/3 
6.  * Version: 1.0.0 
7.  */
8. publicabstractclass ForkJoinService<T> extends RecursiveTask<T>{ 
9.     @Override
10.     protectedabstract T compute(); 
11. } 

2. 定義基類

1. import java.util.List; 
2. import java.util.stream.Collectors; 
3. /** 
4.  * Description: ForkJoin基類 
5.  * Designer: jack 
6.  * Date: 2017/8/3 
7.  * Version: 1.0.0 
8.  */
9. publicclass ForkJoinTest extends ForkJoinService<List<String>> { 
10.     privatestatic ForkJoinTest forkJoinTest; 
11.     privateint threshold;  //閾值
12.     private List<String> list; //待拆分List
13.     private ForkJoinTest(List<String> list, int threshold) { 
14.         this.list = list; 
15.         this.threshold = threshold; 
16.     } 
17.     @Override
18.     protected List<String> compute() { 
19.         // 當end與start之間的差小於閾值時，開始進行實際篩選
20.         if (list.size() < threshold) { 
21.             return list.parallelStream().filter(s -> s.contains("a")).collect(Collectors.toList()); 
22.         } else { 
23.             // 如果當end與start之間的差大於閾值時,將大任務分解成兩個小任務。
24.             int middle = list.size() / 2; 
25.             List<String> leftList = list.subList(0, middle); 
26.             List<String> rightList = list.subList(middle, list.size()); 
27.             ForkJoinTest left = new ForkJoinTest(leftList, threshold); 
28.             ForkJoinTest right = new ForkJoinTest(rightList, threshold); 
29.             // 並行執行兩個“小任務”
30.             left.fork(); 
31.             right.fork(); 
32.             // 把兩個“小任務”的結果合併起來
33.             List<String> join = left.join(); 
34.             join.addAll(right.join()); 
35.             return join; 
36.         } 
37.     } 
38.     /** 
39.      * 獲取ForkJoinTest例項 
40.      * @param list  待處理List 
41.      * @param threshold 閾值 
42.      * @return ForkJoinTest例項 
43.      */
44.     publicstatic ForkJoinService<List<String>> getInstance(List<String> list, int threshold) { 
45.         if (forkJoinTest == null) { 
46.             synchronized (ForkJoinTest.class) { 
47.                 if (forkJoinTest == null) { 
48.                     forkJoinTest = new ForkJoinTest(list, threshold); 
49.                 } 
50.             } 
51.         } 
52.         return forkJoinTest; 
53.     } 
54. } 

3. 執行類 

1. import java.util.ArrayList; 
2. import java.util.Arrays; 
3. import java.util.List; 
4. import java.util.concurrent.ExecutionException; 
5. import java.util.concurrent.ForkJoinPool; 
6. import java.util.concurrent.ForkJoinTask; 
7. /** 
8.  * Description: Fork/Join執行類 
9.  * Designer: jack 
10.  * Date: 2017/8/3 
11.  * Version: 1.0.0 
12.  */
13. publicclass Test { 
14.     publicstaticvoid main(String args[]) throws ExecutionException, InterruptedException { 
15.         String[] strings = {"a", "ah", "b", "ba", "ab", "ac", "sd", "fd", "ar", "te", "se", "te", 
16.                 "sdr", "gdf", "df", "fg", "gh", "oa", "ah", "qwe", "re", "ty", "ui"}; 
17.         List<String> stringList = new ArrayList<>(Arrays.asList(strings)); 
18.         ForkJoinPool pool = new ForkJoinPool(); 
19.         ForkJoinService<List<String>> forkJoinService = ForkJoinTest.getInstance(stringList, 20); 
20.         // 提交可分解的ForkJoinTask任務
21.         ForkJoinTask<List<String>> future = pool.submit(forkJoinService); 
22.         System.out.println(future.get()); 
23.         // 關閉執行緒池
24.         pool.shutdown(); 
25.     } 
26. }