java 線程池 ThreadPoolExecutor

 

1、簡介

 

線程池類爲 Java.util.concurrent.ThreadPoolExecutor，經常使用構造方法爲：

 

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,

long keepAliveTime, TimeUnit unit,

BlockingQueue<Runnable> workQueue,

RejectedExecutionHandler handler)

 

 

corePoolSize： 線程池維護線程的最少數量

maximumPoolSize：線程池維護線程的最大數量

keepAliveTime： 線程池維護線程所容許的空閒時間

unit： 線程池維護線程所容許的空閒時間的單位

workQueue： 線程池所使用的緩衝隊列

handler： 線程池對拒絕任務的處理策略

 

一個任務經過 execute(Runnable)方法被添加到線程池，任務就是一個 Runnable類型的對象，任務的執行方法就是Runnable類型對象的run()方法。

 

先看一副圖，描述了ThreadPoolExecutor的工做機制: 

 

整個ThreadPoolExecutor的任務處理有4步操做：

 

• 第一步，初始的poolSize < corePoolSize，提交的runnable任務，會直接作爲new一個Thread的參數，立馬執行
• 第二步，當提交的任務數超過了corePoolSize，就進入了第二步操做。會將當前的runable提交到一個block queue中
• 第三步，若是block queue是個有界隊列，當隊列滿了以後就進入了第三步。若是poolSize < maximumPoolsize時，會嘗試new 一個Thread的進行救急處理，立馬執行對應的runnable任務
• 第四步，若是第三步救急方案也沒法處理了，就會走到第四步執行reject操做。

幾點說明：(相信這些網上一搜一大把，我這裏簡單介紹下，爲後面作一下鋪墊)

• block queue有如下幾種實現：
  1. ArrayBlockingQueue :  有界的數組隊列
  2. LinkedBlockingQueue : 可支持有界/無界的隊列，使用鏈表實現
  3. PriorityBlockingQueue : 優先隊列，能夠針對任務排序
  4. SynchronousQueue : 隊列長度爲1的隊列，和Array有點區別就是：client thread提交到block queue會是一個阻塞過程，直到有一個worker thread鏈接上來poll task。
• RejectExecutionHandler是針對任務沒法處理時的一些自保護處理：
  1. Reject 直接拋出Reject exception
  2. Discard 直接忽略該runnable，不可取
  3. DiscardOldest 丟棄最先入隊列的的任務
  4. CallsRun 直接讓原先的client thread作爲worker線程，進行執行

 

容易被人忽略的點：

1.  pool threads啓動後，之後的任務獲取都會經過block queue中，獲取堆積的runnable task.

 

因此建議： block size >= corePoolSize ，否則線程池就沒任何意義

2.  corePoolSize 和 maximumPoolSize的區別， 和你們正常理解的數據庫鏈接池不太同樣。

  *  據dbcp pool爲例，會有minIdle , maxActive配置。minIdle表明是常駐內存中的threads數量，maxActive表明是工做的最大線程數。

  *  這裏的corePoolSize就是鏈接池的maxActive的概念，它沒有minIdle的概念(每一個線程能夠設置keepAliveTime，超過多少時間多有任務後銷燬線程，但不會固定保持必定數量的threads)。 

  * 這裏的maximumPoolSize，是一種救急措施的第一層。當threadPoolExecutor的工做threads存在滿負荷，而且block queue隊列也滿了，這時表明接近崩潰邊緣。這時容許臨時起一批threads，用來處理runnable，處理完後立馬退出。

 

因此建議：  maximumPoolSize >= corePoolSize =指望的最大線程數。 (我曾經配置了corePoolSize=1, maximumPoolSize=20, blockqueue爲無界隊列，最後就成了單線程工做的pool。典型的配置錯誤)

 

3. 善用blockqueue和reject組合. 這裏要重點推薦下CallsRun的Rejected Handler，從字面意思就是讓調用者本身來運行。

咱們常常會在線上使用一些線程池作異步處理，好比我前面作的(業務層)異步並行加載技術分析和設計, 將本來串行的請求都變爲了並行操做，但過多的並行會增長系統的負載(好比軟中斷，上下文切換)。因此確定須要對線程池作一個size限制。可是爲了引入異步操做後，避免因在block queue的等待時間過長，因此須要在隊列滿的時，執行一個callsRun的策略，並行的操做又轉爲一個串行處理，這樣就能夠保證儘可能少的延遲影響。

 

因此建議：  RejectExecutionHandler = CallsRun ,  blockqueue size = 2 * poolSize (爲啥是2倍poolSize，主要一個考慮就是瞬間高峯處理，容許一個thread等待一個runnable任務)

當一個任務經過execute(Runnable)方法欲添加到線程池時：

 

l  若是此時線程池中的數量小於corePoolSize，即便線程池中的線程都處於空閒狀態，也要建立新的線程來處理被添加的任務。

l  若是此時線程池中的數量等於 corePoolSize，可是緩衝隊列 workQueue未滿，那麼任務被放入緩衝隊列。

l  若是此時線程池中的數量大於corePoolSize，緩衝隊列workQueue滿，而且線程池中的數量小於maximumPoolSize，建新的線程來處理被添加的任務。

l  若是此時線程池中的數量大於corePoolSize，緩衝隊列workQueue滿，而且線程池中的數量等於maximumPoolSize，那麼經過 handler所指定的策略來處理此任務。也就是：處理任務的優先級爲：核心線程corePoolSize、任務隊列workQueue、最大線程maximumPoolSize，若是三者都滿了，使用handler處理被拒絕的任務。

l  當線程池中的線程數量大於 corePoolSize時，若是某線程空閒時間超過keepAliveTime，線程將被終止。這樣，線程池能夠動態的調整池中的線程數。

 

unit可選的參數爲java.util.concurrent.TimeUnit中的幾個靜態屬性：

NANOSECONDS、

MICROSECONDS、

MILLISECONDS、

SECONDS。

 

workQueue經常使用的是：java.util.concurrent.ArrayBlockingQueue

 

handler有四個選擇：

ThreadPoolExecutor.AbortPolicy()

拋出java.util.concurrent.RejectedExecutionException異常

 

ThreadPoolExecutor.CallerRunsPolicy()

當拋出RejectedExecutionException異常時，會調用rejectedExecution方法

(若是主線程沒有關閉，則主線程調用run方法,源碼以下

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }

)

 

ThreadPoolExecutor.DiscardOldestPolicy()

拋棄舊的任務

 

ThreadPoolExecutor.DiscardPolicy()

拋棄當前的任務

 

2、相關參考

 

一個 ExecutorService，它使用可能的幾個池線程之一執行每一個提交的任務，一般使用 Executors 工廠方法配置。

 

線程池能夠解決兩個不一樣問題：因爲減小了每一個任務調用的開銷，它們一般能夠在執行大量異步任務時提供加強的性能，而且還能夠提供綁定和管理資源（包括執行集合任務時使用的線程）的方法。每一個ThreadPoolExecutor 還維護着一些基本的統計數據，如完成的任務數。

 

爲了便於跨大量上下文使用，此類提供了不少可調整的參數和擴展掛鉤。可是，強烈建議程序員使用較爲方便的 Executors 工廠方法 Executors.newCachedThreadPool()（無界線程池，能夠進行自動線程回收）、Executors.newFixedThreadPool(int)（固定大小線程池）和 Executors.newSingleThreadExecutor()（單個後臺線程），它們均爲大多數使用場景預約義了設置。不然，在手動配置和調整此類時，使用如下指導：

 

核心和最大池大小

ThreadPoolExecutor 將根據 corePoolSize（參見 getCorePoolSize()）和 maximumPoolSize（參見getMaximumPoolSize()）設置的邊界自動調整池大小。當新任務在方法 execute(java.lang.Runnable) 中提交時，若是運行的線程少於 corePoolSize，則建立新線程來處理請求，即便其餘輔助線程是空閒的。若是運行的線程多於corePoolSize 而少於 maximumPoolSize，則僅當隊列滿時才建立新線程。若是設置的 corePoolSize 和 maximumPoolSize相同，則建立了固定大小的線程池。若是將 maximumPoolSize 設置爲基本的無界值（如 Integer.MAX_VALUE），則容許池適應任意數量的併發任務。在大多數狀況下，核心和最大池大小僅基於構造來設置，不過也可使用setCorePoolSize(int) 和 setMaximumPoolSize(int) 進行動態更改