轉(zhuǎn)自： Java 并發(fā)編程筆記：如何使用 ForkJoinPool 以及原理
前言
Java 1.7 引入了一種新的并發(fā)框架—— Fork/Join Framework。
本文的主要目的是介紹 ForkJoinPool 的適用場景，實現(xiàn)原理察郁，以及示例代碼袜硫。
TLDR; 如果覺得文章太長的話，以下就是結(jié)論：
-
ForkJoinPool不是為了替代ExecutorService，而是它的補充合瓢，在某些應(yīng)用場景下性能比ExecutorService更好市俊。（見 Java Tip: When to use ForkJoinPool vs ExecutorService ） -
ForkJoinPool主要用于實現(xiàn)“分而治之”的算法杨凑，特別是分治之后遞歸調(diào)用的函數(shù)，例如 quick sort 等秕衙。 -
ForkJoinPool最適合的是計算密集型的任務(wù)蠢甲，如果存在 I/O，線程間同步据忘，sleep()等會造成線程長時間阻塞的情況時鹦牛，最好配合使用ManagedBlocker。
一個任務(wù)會join其他小任務(wù)勇吊，如果一個小任務(wù)耗時比較長曼追，會造成很多大任務(wù)線程等待。
使用
首先介紹的是大家最關(guān)心的 Fork/Join Framework 的使用方法汉规，如果對使用方法已經(jīng)很熟悉的話礼殊，可以跳過這一節(jié)，直接閱讀原理针史。
用一個特別簡單的求整數(shù)數(shù)組所有元素之和來作為我們現(xiàn)在需要解決的問題吧晶伦。
問題
計算1至1000的正整數(shù)之和。
解決方法
For-loop
最簡單的啄枕，顯然是不使用任何并行編程的手段婚陪，只用最直白的 for-loop 來實現(xiàn)。下面就是具體的實現(xiàn)代碼频祝。
不過為了便于橫向?qū)Ρ让诓危矠榱俗尨a更加 Java Style脆淹，首先我們先定義一個 interface。
public interface Calculator {
long sumUp(long[] numbers);
}
這個 interface 非常簡單沽一，只有一個函數(shù) sumUp盖溺，就是返回數(shù)組內(nèi)所有元素的和。
再寫一個 main 方法铣缠。
public class Main {
public static void main(String[] args) {
long[] numbers = LongStream.rangeClosed(1, 1000).toArray();
Calculator calculator = new MyCalculator();
System.out.println(calculator.sumUp(numbers)); // 打印結(jié)果500500
}
}
接下來就是我們的 Plain Old For-loop Calculator烘嘱，簡稱 POFLC 的實現(xiàn)了。（這其實是個段子蝗蛙，和主題完全無關(guān)拙友，感興趣的請見文末的彩蛋）
public class ForLoopCalculator implements Calculator {
public long sumUp(long[] numbers) {
long total = 0;
for (long i : numbers) {
total += i;
}
return total;
}
}
這段代碼毫無出奇之處，也就不多解釋了歼郭，直接跳入下一節(jié)——并行計算遗契。
ExecutorService
在 Java 1.5 引入 ExecutorService 之后，基本上已經(jīng)不推薦直接創(chuàng)建 Thread 對象病曾，而是統(tǒng)一使用 ExecutorService牍蜂。畢竟從接口的易用程度上來說 ExecutorService 就遠勝于原始的 Thread，更不用提 java.util.concurrent 提供的數(shù)種線程池泰涂，F(xiàn)uture 類鲫竞，Lock 類等各種便利工具。
使用 ExecutorService 的實現(xiàn)
public class ExecutorServiceCalculator implements Calculator {
private int parallism;
private ExecutorService pool;
public ExecutorServiceCalculator() {
parallism = Runtime.getRuntime().availableProcessors(); // CPU的核心數(shù)
pool = Executors.newFixedThreadPool(parallism);
}
private static class SumTask implements Callable<Long> {
private long[] numbers;
private int from;
private int to;
public SumTask(long[] numbers, int from, int to) {
this.numbers = numbers;
this.from = from;
this.to = to;
}
@Override
public Long call() throws Exception {
long total = 0;
for (int i = from; i <= to; i++) {
total += numbers[i];
}
return total;
}
}
@Override
public long sumUp(long[] numbers) {
List<Future<Long>> results = new ArrayList<>();
// 把任務(wù)分解為 n 份逼蒙，交給 n 個線程處理
int part = numbers.length / parallism;
for (int i = 0; i < parallism; i++) {
int from = i * part;
int to = (i == parallism - 1) ? numbers.length - 1 : (i + 1) * part - 1;
results.add(pool.submit(new SumTask(numbers, from, to)));
}
// 把每個線程的結(jié)果相加从绘，得到最終結(jié)果
long total = 0L;
for (Future<Long> f : results) {
try {
total += f.get();
} catch (Exception ignore) {}
}
return total;
}
}
如果對 ExecutorService 不太熟悉的話，推薦閱讀《七天七并發(fā)模型》的第二章是牢，對 Java 的多線程編程基礎(chǔ)講解得比較清晰僵井。當(dāng)然著名的《Java并發(fā)編程實戰(zhàn)》也是不可多得的好書。
ForkJoinPool
前面花了點時間講解了 ForkJoinPool 之前的實現(xiàn)方法驳棱，主要為了在代碼的編寫難度上進行一下對比∨玻現(xiàn)在就列出本篇文章的重點——ForkJoinPool 的實現(xiàn)方法。
public class ForkJoinCalculator implements Calculator {
private ForkJoinPool pool;
private static class SumTask extends RecursiveTask<Long> {
private long[] numbers;
private int from;
private int to;
public SumTask(long[] numbers, int from, int to) {
this.numbers = numbers;
this.from = from;
this.to = to;
}
@Override
protected Long compute() {
// 當(dāng)需要計算的數(shù)字小于6時社搅，直接計算結(jié)果
if (to - from < 6) {
long total = 0;
for (int i = from; i <= to; i++) {
total += numbers[i];
}
return total;
// 否則驻债，把任務(wù)一分為二，遞歸計算
} else {
int middle = (from + to) / 2;
SumTask taskLeft = new SumTask(numbers, from, middle);
SumTask taskRight = new SumTask(numbers, middle+1, to);
taskLeft.fork();
taskRight.fork();
return taskLeft.join() + taskRight.join();
}
}
}
public ForkJoinCalculator() {
// 也可以使用公用的 ForkJoinPool：
// pool = ForkJoinPool.commonPool()
pool = new ForkJoinPool();
}
@Override
public long sumUp(long[] numbers) {
return pool.invoke(new SumTask(numbers, 0, numbers.length-1));
}
}
可以看出形葬，使用了 ForkJoinPool 的實現(xiàn)邏輯全部集中在了 compute() 這個函數(shù)里合呐，僅用了14行就實現(xiàn)了完整的計算過程。特別是笙以，在這段代碼里沒有顯式地“把任務(wù)分配給線程”淌实，只是分解了任務(wù)，而把具體的任務(wù)到線程的映射交給了 ForkJoinPool 來完成。
原理
如果你除了 ForkJoinPool 的用法以外翩伪，對 ForkJoinPoll 的原理也感興趣的話，那么請接著閱讀這一節(jié)谈息。在這一節(jié)中缘屹，我會結(jié)合 ForkJoinPool 的作者 Doug Lea 的論文——《A Java Fork/Join Framework》，盡可能通俗地解釋 Fork/Join Framework 的原理侠仇。
我一直以為轻姿，要理解一樣?xùn)|西的原理，最好就是自己嘗試著去實現(xiàn)一遍逻炊。根據(jù)上面的示例代碼互亮，可以看出 fork() 和 join() 是 Fork/Join Framework “魔法”的關(guān)鍵。我們可以根據(jù)函數(shù)名假設(shè)一下 fork() 和 join() 的作用：
-
fork()：開啟一個新線程（或是重用線程池內(nèi)的空閑線程）余素，將任務(wù)交給該線程處理豹休。 -
join()：等待該任務(wù)的處理線程處理完畢，獲得返回值桨吊。
以上模型似乎可以（威根？）解釋 ForkJoinPool 能夠多線程執(zhí)行的事實，但有一個很明顯的問題
當(dāng)任務(wù)分解得越來越細時视乐，所需要的線程數(shù)就會越來越多洛搀，而且大部分線程處于等待狀態(tài)。
但是如果我們在上面的示例代碼加入以下代碼
System.out.println(pool.getPoolSize());
這會顯示當(dāng)前線程池的大小佑淀，在我的機器上這個值是4留美，也就是說只有4個工作線程。甚至即使我們在初始化 pool 時指定所使用的線程數(shù)為1時伸刃，上述程序也沒有任何問題——除了變成了一個串行程序以外谎砾。
public ForkJoinCalculator() {
pool = new ForkJoinPool(1);
}
這個矛盾可以導(dǎo)出，我們的假設(shè)是錯誤的捧颅，并不是每個 fork() 都會促成一個新線程被創(chuàng)建棺榔，而每個 join() 也不是一定會造成線程被阻塞。Fork/Join Framework 的實現(xiàn)算法并不是那么“顯然”隘道，而是一個更加復(fù)雜的算法——這個算法的名字就叫做 work stealing 算法症歇。
work stealing 算法在 Doung Lea 的論文中有詳細的描述，以下是我在結(jié)合 Java 1.8 代碼的閱讀以后——現(xiàn)有代碼的實現(xiàn)有一部分相比于論文中的描述發(fā)生了變化——得到的相對通俗的解釋：
基本思想
-
ForkJoinPool的每個工作線程都維護著一個工作隊列（WorkQueue）谭梗，這是一個雙端隊列（Deque）忘晤，里面存放的對象是任務(wù)（ForkJoinTask）。 - 每個工作線程在運行中產(chǎn)生新的任務(wù)（通常是因為調(diào)用了
fork()）時激捏，會放入工作隊列的隊尾设塔，并且工作線程在處理自己的工作隊列時，使用的是 LIFO 方式远舅，也就是說每次從隊尾取出任務(wù)來執(zhí)行闰蛔。 - 每個工作線程在處理自己的工作隊列同時痕钢，會嘗試竊取一個任務(wù)（或是來自于剛剛提交到 pool 的任務(wù)，或是來自于其他工作線程的工作隊列）序六，竊取的任務(wù)位于其他線程的工作隊列的隊首任连，也就是說工作線程在竊取其他工作線程的任務(wù)時，使用的是 FIFO 方式例诀。
- 在遇到
join()時随抠，如果需要 join 的任務(wù)尚未完成，則會先處理其他任務(wù)繁涂，并等待其完成拱她。 - 在既沒有自己的任務(wù)，也沒有可以竊取的任務(wù)時扔罪，進入休眠秉沼。
下面來介紹一下關(guān)鍵的兩個函數(shù)：fork() 和 join() 的實現(xiàn)細節(jié)，相比來說 fork() 比 join() 簡單很多矿酵，所以先來介紹 fork()氧猬。
fork
fork() 做的工作只有一件事，既是把任務(wù)推入當(dāng)前工作線程的工作隊列里坏瘩≈迅В可以參看以下的源代碼：
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;
}
join
join() 的工作則復(fù)雜得多，也是 join() 可以使得線程免于被阻塞的原因——不像同名的 Thread.join()倔矾。
- 檢查調(diào)用
join()的線程是否是 ForkJoinThread 線程妄均。如果不是（例如 main 線程），則阻塞當(dāng)前線程哪自，等待任務(wù)完成丰包。如果是，則不阻塞壤巷。 - 查看任務(wù)的完成狀態(tài)邑彪，如果已經(jīng)完成，直接返回結(jié)果胧华。
- 如果任務(wù)尚未完成寄症，但處于自己的工作隊列內(nèi)，則完成它矩动。
- 如果任務(wù)已經(jīng)被其他的工作線程偷走有巧，則竊取這個小偷的工作隊列內(nèi)的任務(wù)（以 FIFO 方式），執(zhí)行悲没，以期幫助它早日完成欲 join 的任務(wù)篮迎。
- 如果偷走任務(wù)的小偷也已經(jīng)把自己的任務(wù)全部做完，正在等待需要 join 的任務(wù)時，則找到小偷的小偷甜橱，幫助它完成它的任務(wù)逊笆。
- 遞歸地執(zhí)行第5步。
將上述流程畫成序列圖的話就是這個樣子：
以上就是 fork() 和 join() 的原理岂傲，這可以解釋 ForkJoinPool 在遞歸過程中的執(zhí)行邏輯难裆，但還有一個問題
最初的任務(wù)是 push 到哪個線程的工作隊列里的？
這就涉及到 submit() 函數(shù)的實現(xiàn)方法了
submit
其實除了前面介紹過的每個工作線程自己擁有的工作隊列以外譬胎，ForkJoinPool 自身也擁有工作隊列，這些工作隊列的作用是用來接收由外部線程（非 ForkJoinThread 線程）提交過來的任務(wù)命锄，而這些工作隊列被稱為 submitting queue 堰乔。
submit() 和 fork() 其實沒有本質(zhì)區(qū)別，只是提交對象變成了 submitting queue 而已（還有一些同步脐恩，初始化的操作）镐侯。submitting queue 和其他 work queue 一樣，是工作線程”竊取“的對象驶冒，因此當(dāng)其中的任務(wù)被一個工作線程成功竊取時苟翻，就意味著提交的任務(wù)真正開始進入執(zhí)行階段。
總結(jié)
在了解了 Fork/Join Framework 的工作原理之后骗污，相信很多使用上的注意事項就可以從原理中找到原因崇猫。例如：為什么在 ForkJoinTask里最好不要存在 I/O 等會阻塞線程的行為狱从？刃鳄，這個我姑且留作思考題吧 :)
還有一些延伸閱讀的內(nèi)容，在此僅提及一下：
-
ForkJoinPool有一個 Async Mode 赂乐，效果是工作線程在處理本地任務(wù)時也使用 FIFO 順序屋厘。這種模式下的ForkJoinPool更接近于是一個消息隊列涕烧，而不是用來處理遞歸式的任務(wù)。 - 在需要阻塞工作線程時汗洒，可以使用
ManagedBlocker议纯。 - Java 1.8 新增加的
CompletableFuture類可以實現(xiàn)類似于 Javascript 的 promise-chain，內(nèi)部就是使用ForkJoinPool來實現(xiàn)的溢谤。 - 采取雙端隊列的一個原因是為了減小競爭（還有其他原因瞻凤？）。
