如何理解Java線程池及其使用方法
多線程的異步執行方式,雖然能夠最大限度發揮多核計算機的計算能力,但是如果不加控制,反而會對系統造成負擔。線程本身也要占用內存空間,大量的線程會占用內存資源并且可能會導致Out of Memory。即便沒有這樣的情況,大量的線程回收也會給GC帶來很大的壓力。
為了避免重復的創建線程,線程池的出現可以讓線程進行復用。通俗點講,當有工作來,就會向線程池拿一個線程,當工作完成后,并不是直接關閉線程,而是將這個線程歸還給線程池供其他任務使用。
接下來從總體到細致的方式,來共同探討線程池。
二、總體的架構來看Executor的框架圖:
接口:Executor,CompletionService,ExecutorService,ScheduledExecutorService
抽象類:AbstractExecutorService
實現類:ExecutorCompletionService,ThreadPoolExecutor,ScheduledThreadPoolExecutor
從圖中就可以看到主要的方法,本文主要討論的是ThreadPoolExecutor
三、研讀ThreadPoolExecutor看一下該類的構造器:
public ThreadPoolExecutor(int paramInt1, int paramInt2, long paramLong, TimeUnit paramTimeUnit,BlockingQueue<Runnable> paramBlockingQueue, ThreadFactory paramThreadFactory,RejectedExecutionHandler paramRejectedExecutionHandler) { this.ctl = new AtomicInteger(ctlOf(-536870912, 0)); this.mainLock = new ReentrantLock(); this.workers = new HashSet(); this.termination = this.mainLock.newCondition(); if ((paramInt1 < 0) || (paramInt2 <= 0) || (paramInt2 < paramInt1) || (paramLong < 0L))throw new IllegalArgumentException(); if ((paramBlockingQueue == null) || (paramThreadFactory == null) || (paramRejectedExecutionHandler == null))throw new NullPointerException(); this.corePoolSize = paramInt1; this.maximumPoolSize = paramInt2; this.workQueue = paramBlockingQueue; this.keepAliveTime = paramTimeUnit.toNanos(paramLong); this.threadFactory = paramThreadFactory; this.handler = paramRejectedExecutionHandler;}
corePoolSize:線程池的核心池大小,在創建線程池之后,線程池默認沒有任何線程。
當有任務過來的時候才會去創建創建線程執行任務。換個說法,線程池創建之后,線程池中的線程數為0,當任務過來就會創建一個線程去執行,直到線程數達到corePoolSize之后,就會被到達的任務放在隊列中。(注意是到達的任務)。換句更精煉的話:corePoolSize表示允許線程池中允許同時運行的最大線程數。
如果執行了線程池的prestartAllCoreThreads()方法,線程池會提前創建并啟動所有核心線程。
maximumPoolSize:線程池允許的最大線程數,他表示最大能創建多少個線程。maximumPoolSize肯定是大于等于corePoolSize。
keepAliveTime:表示線程沒有任務時最多保持多久然后停止。默認情況下,只有線程池中線程數大于corePoolSize時,keepAliveTime才會起作用。換句話說,當線程池中的線程數大于corePoolSize,并且一個線程空閑時間達到了keepAliveTime,那么就是shutdown。
Unit:keepAliveTime的單位。
workQueue:一個阻塞隊列,用來存儲等待執行的任務,當線程池中的線程數超過它的corePoolSize的時候,線程會進入阻塞隊列進行阻塞等待。通過workQueue,線程池實現了阻塞功能
threadFactory:線程工廠,用來創建線程。
handler:表示當拒絕處理任務時的策略。
3.1、任務緩存隊列在前面我們多次提到了任務緩存隊列,即workQueue,它用來存放等待執行的任務。
workQueue的類型為BlockingQueue<Runnable>,通常可以取下面三種類型:
1)有界任務隊列ArrayBlockingQueue:基于數組的先進先出隊列,此隊列創建時必須指定大小;
2)無界任務隊列LinkedBlockingQueue:基于鏈表的先進先出隊列,如果創建時沒有指定此隊列大小,則默認為Integer.MAX_VALUE;
3)直接提交隊列synchronousQueue:這個隊列比較特殊,它不會保存提交的任務,而是將直接新建一個線程來執行新來的任務。
3.2、拒絕策略AbortPolicy:丟棄任務并拋出RejectedExecutionException
CallerRunsPolicy:只要線程池未關閉,該策略直接在調用者線程中,運行當前被丟棄的任務。顯然這樣做不會真的丟棄任務,但是,任務提交線程的性能極有可能會急劇下降。
DiscardOldestPolicy:丟棄隊列中最老的一個請求,也就是即將被執行的一個任務,并嘗試再次提交當前任務。
DiscardPolicy:丟棄任務,不做任何處理。
3.3、線程池的任務處理策略如果當前線程池中的線程數目小于corePoolSize,則每來一個任務,就會創建一個線程去執行這個任務;
如果當前線程池中的線程數目>=corePoolSize,則每來一個任務,會嘗試將其添加到任務緩存隊列當中,若添加成功,則該任務會等待空閑線程將其取出去執行;若添加失敗(一般來說是任務緩存隊列已滿),則會嘗試創建新的線程去執行這個任務;如果當前線程池中的線程數目達到maximumPoolSize,則會采取任務拒絕策略進行處理;
如果線程池中的線程數量大于 corePoolSize時,如果某線程空閑時間超過keepAliveTime,線程將被終止,直至線程池中的線程數目不大于corePoolSize;如果允許為核心池中的線程設置存活時間,那么核心池中的線程空閑時間超過keepAliveTime,線程也會被終止。
3.4、線程池的關閉ThreadPoolExecutor提供了兩個方法,用于線程池的關閉,分別是shutdown()和shutdownNow(),其中:
shutdown():不會立即終止線程池,而是要等所有任務緩存隊列中的任務都執行完后才終止,但再也不會接受新的任務
shutdownNow():立即終止線程池,并嘗試打斷正在執行的任務,并且清空任務緩存隊列,返回尚未執行的任務
3.5、源碼分析首先來看最核心的execute方法,這個方法在AbstractExecutorService中并沒有實現,從Executor接口,直到ThreadPoolExecutor才實現了改方法,
ExecutorService中的submit(),invokeAll(),invokeAny()都是調用的execute方法,所以execute是核心中的核心,源碼分析將圍繞它逐步展開。
public void execute(Runnable command) {if (command == null) throw new NullPointerException();/* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn’t, by returning false. * 如果正在運行的線程數小于corePoolSize,那么將調用addWorker 方法來創建一個新的線程,并將該任務作為新線程的第一個任務來執行。 當然,在創建線程之前會做原子性質的檢查,如果條件不允許,則不創建線程來執行任務,并返回false. * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * 如果一個任務成功進入阻塞隊列,那么我們需要進行一個雙重檢查來確保是我們已經添加一個線程(因為存在著一些線程在上次檢查后他已經死亡)或者 當我們進入該方法時,該線程池已經關閉。所以,我們將重新檢查狀態,線程池關閉的情況下則回滾入隊列,線程池沒有線程的情況則創建一個新的線程。 * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. 如果任務無法入隊列(隊列滿了),那么我們將嘗試新開啟一個線程(從corepoolsize到擴充到maximum),如果失敗了,那么可以確定原因,要么是 線程池關閉了或者飽和了(達到maximum),所以我們執行拒絕策略。 */// 1.當前線程數量小于corePoolSize,則創建并啟動線程。int c = ctl.get();if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true))// 成功,則返回return; c = ctl.get();}// 2.步驟1失敗,則嘗試進入阻塞隊列,if (isRunning(c) && workQueue.offer(command)) { // 入隊列成功,檢查線程池狀態,如果狀態部署RUNNING而且remove成功,則拒絕任務 int recheck = ctl.get(); if (! isRunning(recheck) && remove(command))reject(command); // 如果當前worker數量為0,通過addWorker(null, false)創建一個線程,其任務為null else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 3. 步驟1和2失敗,則嘗試將線程池的數量有corePoolSize擴充至maxPoolSize,如果失敗,則拒絕任務else if (!addWorker(command, false)) reject(command); }
相信看了代碼也是一臉懵,接下來用一個流程圖來講一講,他究竟干了什么事:
結合上面的流程圖來逐行解析,首先前面進行空指針檢查,
wonrkerCountOf()方法能夠取得當前線程池中的線程的總數,取得當前線程數與核心池大小比較,
如果小于,將通過addWorker()方法調度執行。 如果大于核心池大小,那么就提交到等待隊列。 如果進入等待隊列失敗,則會將任務直接提交給線程池。 如果線程數達到最大線程數,那么就提交失敗,執行拒絕策略。excute()方法中添加任務的方式是使用addWorker()方法,看一下源碼,一起學習一下。
private boolean addWorker(Runnable firstTask, boolean core) {retry: // 外層循環,用于判斷線程池狀態for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN &&! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))return false; // 內層的循環,任務是將worker數量加1 for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false;if (compareAndIncrementWorkerCount(c)) break retry;c = ctl.get(); // Re-read ctlif (runStateOf(c) != rs) continue retry;// else CAS failed due to workerCount change; retry inner loop }}// worker加1后,接下來將woker添加到HashSet<Worker>中,并啟動workerboolean workerStarted = false;boolean workerAdded = false;Worker w = null;try { final ReentrantLock mainLock = this.mainLock; w = new Worker(firstTask); final Thread t = w.thread; if (t != null) {mainLock.lock();try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int c = ctl.get(); int rs = runStateOf(c); 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();} // 如果往HashSet<Worker>添加成功,則啟動該線程if (workerAdded) { t.start(); workerStarted = true;} }} finally { if (! workerStarted)addWorkerFailed(w);}return workerStarted; }
addWorker(Runnable firstTask, boolean core)的主要任務是創建并啟動線程。
他會根據當前線程的狀態和給定的值(core or maximum)來判斷是否可以創建一個線程。
addWorker共有四種傳參方式。execute使用了其中三種,分別為:
1.addWorker(paramRunnable, true)
線程數小于corePoolSize時,放一個需要處理的task進Workers Set。如果Workers Set長度超過corePoolSize,就返回false.
2.addWorker(null, false)
放入一個空的task進workers Set,長度限制是maximumPoolSize。這樣一個task為空的worker在線程執行的時候會去任務隊列里拿任務,這樣就相當于創建了一個新的線程,只是沒有馬上分配任務。
3.addWorker(paramRunnable, false)
當隊列被放滿時,就嘗試將這個新來的task直接放入Workers Set,而此時Workers Set的長度限制是maximumPoolSize。如果線程池也滿了的話就返回false.
還有一種情況是execute()方法沒有使用的
addWorker(null, true)
這個方法就是放一個null的task進Workers Set,而且是在小于corePoolSize時,如果此時Set中的數量已經達到corePoolSize那就返回false,什么也不干。實際使用中是在prestartAllCoreThreads()方法,這個方法用來為線程池預先啟動corePoolSize個worker等待從workQueue中獲取任務執行。
執行流程:
1、判斷線程池當前是否為可以添加worker線程的狀態,可以則繼續下一步,不可以return false:
A、線程池狀態>shutdown,可能為stop、tidying、terminated,不能添加worker線程
B、線程池狀態==shutdown,firstTask不為空,不能添加worker線程,因為shutdown狀態的線程池不接收新任務
C、線程池狀態==shutdown,firstTask==null,workQueue為空,不能添加worker線程,因為firstTask為空是為了添加一個沒有任務的線程再從workQueue獲取task,而workQueue為空,說明添加無任務線程已經沒有意義
2、線程池當前線程數量是否超過上限(corePoolSize 或 maximumPoolSize),超過了return false,沒超過則對workerCount+1,繼續下一步
3、在線程池的ReentrantLock保證下,向Workers Set中添加新創建的worker實例,添加完成后解鎖,并啟動worker線程,如果這一切都成功了,return true,如果添加worker入Set失敗或啟動失敗,調用addWorkerFailed()邏輯
四、常見的四種線程池4.1、newFixedThreadPoolpublic static ExecutorService newFixedThreadPool(int var0) { return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());}public static ExecutorService newFixedThreadPool(int var0, ThreadFactory var1) {return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), var1);}
固定大小的線程池,可以指定線程池的大小,該線程池corePoolSize和maximumPoolSize相等,阻塞隊列使用的是LinkedBlockingQueue,大小為整數最大值。
該線程池中的線程數量始終不變,當有新任務提交時,線程池中有空閑線程則會立即執行,如果沒有,則會暫存到阻塞隊列。對于固定大小的線程池,不存在線程數量的變化。同時使用無界的LinkedBlockingQueue來存放執行的任務。當任務提交十分頻繁的時候,LinkedBlockingQueue
迅速增大,存在著耗盡系統資源的問題。而且在線程池空閑時,即線程池中沒有可運行任務時,它也不會釋放工作線程,還會占用一定的系統資源,需要shutdown。
4.2、newSingleThreadExecutorpublic static ExecutorService newSingleThreadExecutor() { return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()));}public static ExecutorService newSingleThreadExecutor(ThreadFactory var0) { return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), var0));}
單個線程線程池,只有一個線程的線程池,阻塞隊列使用的是LinkedBlockingQueue,若有多余的任務提交到線程池中,則會被暫存到阻塞隊列,待空閑時再去執行。按照先入先出的順序執行任務。
4.3、newCachedThreadPoolpublic static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue());}public static ExecutorService newCachedThreadPool(ThreadFactory var0) { return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue(), var0);}
緩存線程池,緩存的線程默認存活60秒。線程的核心池corePoolSize大小為0,核心池最大為Integer.MAX_VALUE,阻塞隊列使用的是SynchronousQueue。是一個直接提交的阻塞隊列, 他總會迫使線程池增加新的線程去執行新的任務。在沒有任務執行時,當線程的空閑時間超過keepAliveTime(60秒),則工作線程將會終止被回收,當提交新任務時,如果沒有空閑線程,則創建新線程執行任務,會導致一定的系統開銷。如果同時又大量任務被提交,而且任務執行的時間不是特別快,那么線程池便會新增出等量的線程池處理任務,這很可能會很快耗盡系統的資源。
4.4、newScheduledThreadPoolpublic static ScheduledExecutorService newScheduledThreadPool(int var0) { return new ScheduledThreadPoolExecutor(var0);}public static ScheduledExecutorService newScheduledThreadPool(int var0, ThreadFactory var1) { return new ScheduledThreadPoolExecutor(var0, var1);}
定時線程池,該線程池可用于周期性地去執行任務,通常用于周期性的同步數據。
scheduleAtFixedRate:是以固定的頻率去執行任務,周期是指每次執行任務成功執行之間的間隔。
schedultWithFixedDelay:是以固定的延時去執行任務,延時是指上一次執行成功之后和下一次開始執行的之前的時間。
五、使用實例5.1、newFixedThreadPool實例public class FixPoolDemo { private static Runnable getThread(final int i) {return new Runnable() { @Override public void run() {try { Thread.sleep(500);} catch (InterruptedException e) { e.printStackTrace();}System.out.println(i); }}; } public static void main(String args[]) {ExecutorService fixPool = Executors.newFixedThreadPool(5);for (int i = 0; i < 10; i++) { fixPool.execute(getThread(i));}fixPool.shutdown(); }}5.2、newCachedThreadPool實例
public class CachePool { private static Runnable getThread(final int i){return new Runnable() { @Override public void run() {try { Thread.sleep(1000);}catch (Exception e){}System.out.println(i); }}; } public static void main(String args[]){ExecutorService cachePool = Executors.newCachedThreadPool();for (int i=1;i<=10;i++){ cachePool.execute(getThread(i));} }}
這里沒用調用shutDown方法,這里可以發現過60秒之后,會自動釋放資源。
5.3、newSingleThreadExecutorpublic class SingPoolDemo { private static Runnable getThread(final int i){return new Runnable() { @Override public void run() {try { Thread.sleep(500);} catch (InterruptedException e) { e.printStackTrace();}System.out.println(i); }}; } public static void main(String args[]) throws InterruptedException {ExecutorService singPool = Executors.newSingleThreadExecutor();for (int i=0;i<10;i++){ singPool.execute(getThread(i));}singPool.shutdown(); }
這里需要注意一點,newSingleThreadExecutor和newFixedThreadPool一樣,在線程池中沒有任務時可執行,也不會釋放系統資源的,所以需要shudown。
5.4、newScheduledThreadPoolpublic class ScheduledExecutorServiceDemo { public static void main(String args[]) {ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);ses.scheduleAtFixedRate(new Runnable() { @Override public void run() {try { Thread.sleep(4000); System.out.println(Thread.currentThread().getId() + '執行了');} catch (InterruptedException e) { e.printStackTrace();} }}, 0, 2, TimeUnit.SECONDS); }}六、總結6.1、如何選擇線程池數量
線程池的大小決定著系統的性能,過大或者過小的線程池數量都無法發揮最優的系統性能。
當然線程池的大小也不需要做的太過于精確,只需要避免過大和過小的情況。一般來說,確定線程池的大小需要考慮CPU的數量,內存大小,任務是計算密集型還是IO密集型等因素
NCPU = CPU的數量
UCPU = 期望對CPU的使用率 0 ≤ UCPU ≤ 1
W/C = 等待時間與計算時間的比率
如果希望處理器達到理想的使用率,那么線程池的最優大小為:
線程池大小=NCPU *UCPU(1+W/C)
在Java中使用
int ncpus = Runtime.getRuntime().availableProcessors();
獲取CPU的數量。
6.2、線程池工廠Executors的線程池如果不指定線程工廠會使用Executors中的DefaultThreadFactory,默認線程池工廠創建的線程都是非守護線程。
使用自定義的線程工廠可以做很多事情,比如可以跟蹤線程池在何時創建了多少線程,也可以自定義線程名稱和優先級。如果將
新建的線程都設置成守護線程,當主線程退出后,將會強制銷毀線程池。
下面這個例子,記錄了線程的創建,并將所有的線程設置成守護線程。
public class ThreadFactoryDemo { public static class MyTask1 implements Runnable{@Overridepublic void run() { System.out.println(System.currentTimeMillis()+'Thrad ID:'+Thread.currentThread().getId()); try {Thread.sleep(100); } catch (InterruptedException e) {e.printStackTrace(); }} } public static void main(String[] args){ MyTask1 task = new MyTask1();ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MICROSECONDS, new SynchronousQueue<Runnable>(), new ThreadFactory() { @Override public Thread newThread(Runnable r) {Thread t = new Thread(r);t.setDaemon(true);System.out.println('創建線程'+t);return t; }});for (int i = 0;i<=4;i++){ es.submit(task);} }}6.3、擴展線程池
ThreadPoolExecutor是可以拓展的,它提供了幾個可以在子類中改寫的方法:beforeExecute,afterExecute和terimated。
在執行任務的線程中將調用beforeExecute和afterExecute,這些方法中還可以添加日志,計時,監視或統計收集的功能,
還可以用來輸出有用的調試信息,幫助系統診斷故障。下面是一個擴展線程池的例子:
public class ThreadFactoryDemo { public static class MyTask1 implements Runnable{@Overridepublic void run() { System.out.println(System.currentTimeMillis()+'Thrad ID:'+Thread.currentThread().getId()); try {Thread.sleep(100); } catch (InterruptedException e) {e.printStackTrace(); }} } public static void main(String[] args){ MyTask1 task = new MyTask1();ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MICROSECONDS, new SynchronousQueue<Runnable>(), new ThreadFactory() { @Override public Thread newThread(Runnable r) {Thread t = new Thread(r);t.setDaemon(true);System.out.println('創建線程'+t);return t; }});for (int i = 0;i<=4;i++){ es.submit(task);} }}6.4、線程池的正確使用
以下阿里編碼規范里面說的一段話:
線程池不允許使用Executors去創建,而是通過ThreadPoolExecutor的方式,這樣的處理方式讓寫的同學更加明確線程池的運行規則,規避資源耗盡的風險。 說明:Executors各個方法的弊端:
1)newFixedThreadPool和newSingleThreadExecutor:主要問題是堆積的請求處理隊列可能會耗費非常大的內存,甚至OOM。
2)newCachedThreadPool和newScheduledThreadPool:主要問題是線程數最大數是Integer.MAX_VALUE,可能會創建數量非常多的線程,甚至OOM。
七、手動創建線程池有幾個注意點7.1、任務獨立如何任務依賴于其他任務,那么可能產生死鎖。例如某個任務等待另一個任務的返回值或執行結果,那么除非線程池足夠大,否則將發生線程饑餓死鎖。
7.2、合理配置阻塞時間過長的任務如果任務阻塞時間過長,那么即使不出現死鎖,線程池的性能也會變得很糟糕。在Java并發包里可阻塞方法都同時定義了限時方式和不限時方式。例如
Thread.join,BlockingQueue.put,CountDownLatch.await等,如果任務超時,則標識任務失敗,然后中止任務或者將任務放回隊列以便隨后執行,這樣,無論任務的最終結果是否成功,這種辦法都能夠保證任務總能繼續執行下去。
7.3、設置合理的線程池大小只需要避免過大或者過小的情況即可,上文的公式線程池大小=NCPU *UCPU(1+W/C)。
7.4、選擇合適的阻塞隊列newFixedThreadPool和newSingleThreadExecutor都使用了無界的阻塞隊列,無界阻塞隊列會有消耗很大的內存,如果使用了有界阻塞隊列,它會規避內存占用過大的問題,但是當任務填滿有界阻塞隊列,新的任務該怎么辦?在使用有界隊列是,需要選擇合適的拒絕策略,隊列的大小和線程池的大小必須一起調節。對于非常大的或者無界的線程池,可以使用SynchronousQueue來避免任務排隊,以直接將任務從生產者提交到工作者線程。
下面是Thrift框架處理socket任務所使用的一個線程池,可以看一下FaceBook的工程師是如何自定義線程池的。
private static ExecutorService createDefaultExecutorService(Args args) { SynchronousQueue executorQueue = new SynchronousQueue(); return new ThreadPoolExecutor(args.minWorkerThreads, args.maxWorkerThreads, 60L, TimeUnit.SECONDS, executorQueue);}
以上就是如何理解Java線程池及其使用方法的詳細內容,更多關于Java線程池的資料請關注好吧啦網其它相關文章!
相關文章: