原文: https://chenmingyu.top/concurrent-threadpool/java
線程池用來處理異步任務或者併發執行的任務編程
優勢:數組
java
中線程池使用ThreadPoolExecutor
實現緩存
ThreadPoolExecutor
提供了四個構造函數,其餘三個構造函數最終調用的都是下面這個構造函數併發
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }
入參:框架
corePoolSize
:線程池的核心線程數量異步
線程池維護的核心線程數量,當線程池初始化後,核心線程數量爲零,當有任務來到的時候纔會建立線程去執行任務,當線程池中的工做線程數量等於核心線程數量時,新到的任務就會放到緩存隊列中函數
maximumPoolSize
:線程池容許建立的最大線程數量oop
當阻塞隊列滿了的時候,而且線程池中建立的線程數量小於maximumPoolSize
,此時會建立新的線程執行任務ui
keepAliveTime
:線程活動保持時間
只有當線程池數量大於核心線程數量時,keepAliveTime
纔會有效,若是當前線程數量大於核心線程數量時,而且線程的空閒時間達到keepAliveTime
,當前線程終止,直到線程池數量等於核心線程數
unit
:線程活動保持時間的單位
keepAliveTime
的單位,包括:TimeUnit.DAYS
天,TimeUnit.HOURS
小時,TimeUnit.MINUTES
分鐘,TimeUnit.SECONDS
秒,TimeUnit.MILLISECONDS
毫秒,TimeUnit.MICROSECONDS
微秒,TimeUnit.NANOSECONDS
納秒
workQueue
:任務隊列,用來保存等待執行任務的阻塞隊列
ArrayBlockingQueue
:是一個基於數組結構的有界隊列
LinkedBlockingQueue
:是一個基於鏈表結構的阻塞隊列
SynchronousQueue
:不存儲元素的阻塞隊列,每個插入操做必須等到下一個線程調用移除操做,不然插入操做一直阻塞
PriorityBlockingQueue
:一個具備優先級的無線阻塞隊列
threadFactory
:用來建立線程的工廠
handler
:飽和策略,當線程池和隊列都滿了的時候,必需要採起一種策略處理新的任務,默認策略是AbortPolicy
,根據本身需求選擇合適的飽和策略
AbortPolicy
:直接拋出異常
CallerRunsPolicy
:用調用者所在的線程來運行當前任務
DiscardOldestPolicy
:丟棄隊列裏面最近的一個任務,並執行當前任務
DiscardPolicy
:不處理,丟棄掉
固然咱們也能夠經過實現RejectedExecutionHandler
去自定義實現處理策略
入參不一樣,線程池的運行機制也不一樣,瞭解每一個入參的含義因爲咱們更透傳的理解線程池的實現原理
線程池處理提交任務流程以下
處理流程:
ThreadPoolExecutor
使用execute(Runnable command)
和submit(Runnable task)
向線程池中提交任務,在submit(Runnable task)
方法中調用了execute(Runnable command)
,因此咱們只要瞭解execute(Runnable command)
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); // 獲取線程池狀態,而且能夠經過ctl獲取到當前線程池數量及線程池狀態 int c = ctl.get(); // 若是工做線程數小於核心線程數量,則建立一個新線程執行任務 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 若是不符合上面條件,當前線程處於運行狀態而且寫入阻塞隊列成功 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 雙重檢查,再次獲取線程狀態,若是當前線程狀態變爲非運行狀態,則從隊列中移除任務,執行拒絕策略 if (! isRunning(recheck) && remove(command)) reject(command); // 檢查工做線程數量是否爲0 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //建立線程執行任務,若是添加失敗則執行拒絕策略 else if (!addWorker(command, false)) reject(command); }
execute(Runnable command)
方法中咱們比較關心的就是如何建立新的線程執行任務,就addWorker(command, true)
方法
workQueue.offer(command)
方法是用來向阻塞隊列中添加任務的
reject(command)
方法會根據建立線程池時傳入的飽和策略對任務進行處理,例如默認的AbortPolicy
,查看源碼後知道就是直接拋了個RejectedExecutionException
異常,其餘的飽和策略的源碼也是特別簡單
關於線程池狀態與工做線程的數量是如何表示的
在ThreadPoolExecutor
中使用一個AtomicInteger
類型變量表示
/** * ctl表示兩個信息,一個是線程池的狀態(高3位表示),一個是當前線程池的數量(低29位表示),這個跟咱們前面 * 說過的讀寫鎖的state變量是同樣的,以一個變量記錄兩個信息,都是以利用int的32個字節,高十六位表述讀,低十 * 六位表示寫鎖 */ private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); //低29位保存線程池數量 private static final int COUNT_BITS = Integer.SIZE - 3; //線程池最大容量 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // 運行狀態存儲在高3位 // 運行狀態 private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS;
addWorker(command, boolean)
建立工做線程,執行任務
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); // 線程池狀態 int rs = runStateOf(c); // 判斷線程池狀態,以及阻塞隊列是否爲空 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { // 獲取線程工做線程數量 int wc = workerCountOf(c); // 判斷是否大於最大容量,以及根據傳入的core判斷是否大於核心線程數量仍是最大線程數量 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 增長工做線程數量 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl //若是線程池狀態改變,則重試 if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 建立Worker,內部建立了一個新的線程 w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); // 線程池狀態判斷 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); // 將建立的線程添加到線程池 workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { //執行任務,首先會執行Worker對象的firstTask t.start(); workerStarted = true; } } } finally { //若是任務執行失敗 if (! workerStarted) //移除worker addWorkerFailed(w); } return workerStarted; }
ThreadPoolExecutor
中關閉線程池使用shutdown()
和shutdownNow()
方法,原理都是經過遍歷線程池中的線程,對線程進行中斷
for (Worker w : workers) { Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) break; }
Executor
框架將任務的提交與任務的執行進行分離
Executors
提供了一系列工廠方法用於創先線程池,返回的線程池都實現了 ExecutorService
接口
工廠方法:
newFixedThreadPool
:用於建立固定數目線程的線程池newCachedThreadPool
:用於建立一個可緩存的線程池,調用execute將重用之前構造的線程,若是現有線程沒有可用的,則建立一個新線 程並添加到池中。終止並從緩存中移除那些已有 60 秒鐘未被使用的線程newSingleThreadExecutor
:用於建立只有一個線程的線程池newScheduledThreadPool
:用於建立一個支持定時及週期性的任務執行的線程池在阿里巴巴手冊中強制要求禁止使用Executors
提供的工廠方法建立線程池
這個確實是一個很嚴重的問題,咱們部門曾經就出現過使用FixedThreadPool
線程池,致使OOM,這是由於線程執行任務的時候被阻塞或耗時很長時間,致使阻塞隊列一直在添加任務,直到內存被打滿,報OOM
因此咱們在使用線程池的時候要使用ThreadPoolExecutor
的構造函數去建立線程池,根據本身的任務類型來肯定核心線程數和最大線程數,選擇適合阻塞隊列和阻塞隊列的長度
合理的配置線程池須要分析一下任務的性質(使用ThreadPoolExecutor
建立線程池):
CPU密集型任務應配置竟可能小的線程,好比 cpu數量+1
IO密集型任務並非一直在執行任務,應該配置儘量多的線程,好比 cpu數量x2
可經過Runtime.getRuntime().availableProcessors()
獲取cpu數量
執行的任務有調用外部接口比較費時的時候,這時cup空閒的時間就越長,能夠將線程池數量設置大一些,這樣cup空閒的時間就能夠去執行別的任務
建議使用有界隊列,可根據須要將長度設置大一些,防止OOM
參考:java併發編程的藝術
推薦閱讀:
java併發編程 | 鎖詳解:AQS,Lock,ReentrantLock,ReentrantReadWriteLock