[Java并發編程（二）] 線程池 FixedThreadPool、CachedThreadPool、ForkJoinPool？為后臺任務選擇合適的 Java executors ...

摘要

Java 和其他平臺相比最大的優勢在于它可以很好的利用資源來進行并行計算。確實，在 JVM 上可以輕而易舉地在后臺執行一段代碼，并在需要使用它的時候消費計算的結果。同時，它也讓開發者可以更好的利用現代計算機硬件所帶來計算能力。

但是，想讓計算正確并不容易，或許對于開發者最大的挑戰是編寫一個總是能運行正確的程序，而不是我們熟悉的 “在我機器上” 是正確的。

這篇文章會看看 Executor 里提供的不同選擇。

正文

Java Executors 詳解

簡言之，Executor 是一個接口，它旨在將任務的聲明與實際計算解耦。

public interface Executor {void execute(Runnable command);}

它以 Runnable 實例的形式接受任務。線程會在某個時間點獲取任務并執行 Runnable::run 方法。但是，真正有難度的通常是如何選擇將要使用的 Executor 實現。在 Executors 類中已經有一些可供使用的默認實現。讓我們來看看它們是什么以及何時選擇。

總體上說，當選擇供后臺計算的 Executor 時，通常可以考慮 3 個主要的問題：

• 默認希望有多少個線程并行執行？
• 當所有可用線程都處于忙碌狀態時，Executor 會如何處理一個提交的任務？（如：通常它要么使用更多的線程或者要么將任務加入到隊列中）
• 是否希望限制任務隊列的大小，如果隊列滿了會怎樣？

如果希望顯式地控制這些問題的答案，可以使用 JDK 提供的靈活的 API 來創建自己的 ThreadPoolExecutor 。ThreadPoolExecutor 的構造器顯式地要求提供問題的答案：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

以下描述了這些參數的含義以及它們是如何回答以上問題的：

• int corePoolSize - 線程池初始啟動時線程的數量
• int maximumPoolSize - 當核心線程繁忙時，希望線程池增長的最大值
• long keepAliveTime, TimeUnit 單位 - 如果線程空閑不再工作，是否希望關閉它？線程池需要等待多久才關閉它？
• BlockingQueue workQueue - 任務如果不能立即處理是如何處理的？是否希望限制任務隊列的大小？
• RejectedExecutionHandler handler - 如果 Executor 不接受任務會怎樣？是否應該拋出異常？調用方是否應該自行處理任務？

下面列出了由工廠方法 Executors 創建的不同 ExecutorServices 的差異。希望對在面臨如上問題并做出選擇時有幫助。

newFixedThreadPool(int nThread) - n 個線程會同時進行處理，當線程池滿后，新的任務會被加入到大小沒有限制的隊列中。比較適合 CPU 密集型的任務。

newWorkStealingPool(int parallelism) - 會更加所需的并行層次來動態創建和關閉線程。它同樣會試圖減少任務隊列的大小，所以比較適于高負載的環境。同樣也比較適用于當執行的任務會創建更多任務，如遞歸任務。

newSingleThreadExecutor() - 創建一個不可配置的 newFixedThreadPool(1)，所以一個線程會執行所有的任務。它適用于明確知道可預測性以及任務需要按順序執行的情況。

newCachedThreadPool() - 不會將任務加入隊列。可以將它看成一個最大值為 0 的隊列。如果當前線程都處于繁忙狀態，它會創建另外一個線程來執行任務。它有時也會重用線程。它適用于防止 DOS 攻擊。緩存線程池的問題在于它不知道該合適停止創建線程。設想需要執行大量計算的任務時，如果將任務提交給 Executor ，更多的線程會消耗更多的 CPU，同時每個任務的執行也會花更長時間。這是個多米諾效應，有更多的任務會被記錄下。這樣越來越多的線程會被創建，而任務的執行會更慢。很難解決這個負反饋環的問題。

所以對于大多數情況， Executors::newFixedThreadPool(int nThreads) 是當我們想要使用線程池時首先考慮的選擇對象。對于計算密集型的任務它通常能提供近于最優的吞吐量，對于 IO 密集型的任務也不會使任何問題變得更糟。至少如果在我們使用這些 Executor 遇到問題并進行性能調優時，不會毫無頭緒。

ForkJoinPool 與 ManagedBlockers

當然 JVM 上有一個默認選擇的 Executor：通用的 ForkJoinPool，它是由 JVM 預設的用來并行處理流以及執行類似 CompletableFuture::supplyAsync 的任務。

聽起來很美？預設的，隨時隨地可用，最先進的線程池。還希望哪些其他的特性？這里有個忠告，如果有件事情聽起來太好了，那么一定需要擦亮眼睛。ForkJoinPool 簡直太好了，除了它是通用的 common ，（即被整個 JVM 共享），它可以被在同一 JVM 進程內的所有、任何組件使用。

如果不小心讓不合適的任務污染了它，可能會讓整個 JVM 進程受到影響。所以如果不小心讓 common 池中的工作線程阻塞，可能是沒有正確地使用它。

讓我們來看看如何讓它變得更好。ForkJoinPool 設計的初衷是為了解決有些任務會阻塞工作線程的情況，所以它提供了處理這種阻塞的 API 。

讓我們歡迎 — ManagedBlocker - 可以用它來給 ForkJoinPool 傳遞信號擴展它的并行能力，從而補償潛在可能被阻塞的工作線程。

假設我們有一個 Call 實例，與 Retrofit 2 Call 類似，它包含所有查詢所需要的 endpoint 信息，以及如何將結果轉換成對象的信息。開始使用 Retrofit 2，盡管這篇文章主要是寫 Android ，但總體的概念與在 JVM 上使用 Retrofit 是一樣的。它提供了一套很好的 HTTP 請求的 API 。

class WS<E> implements ManagedBlocker {private final Call<E> call;volatile E item = null;public WS(Call<E> call) {this.call = call;}public boolean block() throws InterruptedException {if (item == null)item = call.execute().body();return true;}public boolean isReleasable() {return item != null;}public E getItem() { // call after pool.managedBlock completesreturn item;}}

現在，當我們想要調用 Call::execute 的時候，我們需要保證它是通過 ForkJoinPool::managedBlock 方法進行調用的

WS ws = new WS(call);ForkJoinPool.managedBlock(ws);ws.getItem(); // obtain the result

顯然，當在 FJP 以外運行的時候它毫無意義，在線程池上運行時才有意義。FJP 會在線程出現阻塞時生成多的工作線程。需要提醒的是，這并不是銀彈，相反，很有可能是錯誤的，因為 ManagedBlocker API 是用來處理可能被阻塞的 synchronizer 對象的。這里我們是在處理一個阻塞網絡調用，它可以處理當我們查詢 4 個 urls FJP 計算資源被耗盡的情況。

總結

本篇文章我看了 Executors 類提供給我們的選擇，以及何時使用各個 Executor 的策略。對于 CPU 密集型的任務，newFixedThreadPool 可以適用大多數場景，除非明確知道另外一個選擇更好。但是，對于 IO 密集型的任務，并不簡單。可以通過將 IO 調用包裝到 ManagedBlocker 里并使用 ForkJoinPool 來增強它內部的并行能力。

參考

zeroturnaround: FixedThreadPool, CachedThreadPool, or ForkJoinPool? Picking correct Java executors for background tasks

結束

轉載于:https://www.cnblogs.com/richaaaard/p/6601445.html

總結

以上是生活随笔為你收集整理的[Java并发编程（二）] 线程池 FixedThreadPool、CachedThreadPool、ForkJoinPool？为后台任务选择合适的 Java executors...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。