線程池的技術(shù)背景
在面向?qū)ο缶幊讨校瑒?chuàng)建和銷毀對象是很費時間的,因為創(chuàng)建一個對象要獲取內(nèi)存資源或者其它更多資源。在Java中更是如此,虛擬機將試圖跟蹤每一個對象,以便能夠在對象銷毀后進(jìn)行垃圾回收。
所以提高服務(wù)程序效率的一個手段就是盡可能減少創(chuàng)建和銷毀對象的次數(shù),特別是一些很耗資源的對象創(chuàng)建和銷毀。如何利用已有對象來服務(wù)就是一個需要解決的關(guān)鍵問題,其實這就是一些”池化資源”技術(shù)產(chǎn)生的原因。
例如Android中常見到的很多通用組件一般都離不開”池”的概念,如各種圖片加載庫,網(wǎng)絡(luò)請求庫,即使Android的消息傳遞機制中的Meaasge當(dāng)使用Meaasge.obtain()就是使用的Meaasge池中的對象,因此這個概念很重要。本文將介紹的線程池技術(shù)同樣符合這一思想。
線程池的優(yōu)點:
1.重用線程池中的線程,減少因?qū)ο髣?chuàng)建,銷毀所帶來的性能開銷;
2.能有效的控制線程的最大并發(fā)數(shù),提高系統(tǒng)資源利用率,同時避免過多的資源競爭,避免堵塞;
3.能夠多線程進(jìn)行簡單的管理,使線程的使用簡單、高效。
線程池框架Executor
java中的線程池是通過Executor框架實現(xiàn)的,Executor 框架包括類:Executor,Executors,ExecutorService,ThreadPoolExecutor ,Callable和Future、FutureTask的使用等。
Executor: 所有線程池的接口,只有一個方法。
1
2
3
|
public interface Executor { void execute(Runnable command); } |
ExecutorService: 增加Executor的行為,是Executor實現(xiàn)類的最直接接口。
Executors: 提供了一系列工廠方法用于創(chuàng)先線程池,返回的線程池都實現(xiàn)了ExecutorService 接口。
ThreadPoolExecutor:線程池的具體實現(xiàn)類,一般用的各種線程池都是基于這個類實現(xiàn)的。 構(gòu)造方法如下:
1
2
3
4
5
6
7
8
9
10
|
public ThreadPoolExecutor( int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this (corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } |
corePoolSize:線程池的核心線程數(shù),線程池中運行的線程數(shù)也永遠(yuǎn)不會超過 corePoolSize 個,默認(rèn)情況下可以一直存活。可以通過設(shè)置allowCoreThreadTimeOut為True,此時 核心線程數(shù)就是0,此時keepAliveTime控制所有線程的超時時間。
maximumPoolSize:線程池允許的最大線程數(shù);
keepAliveTime: 指的是空閑線程結(jié)束的超時時間;
unit :是一個枚舉,表示 keepAliveTime 的單位;
workQueue:表示存放任務(wù)的BlockingQueue<Runnable隊列。
BlockingQueue:阻塞隊列(BlockingQueue)是java.util.concurrent下的主要用來控制線程同步的工具。如果BlockQueue是空的,從BlockingQueue取東西的操作將會被阻斷進(jìn)入等待狀態(tài),直到BlockingQueue進(jìn)了東西才會被喚醒。同樣,如果BlockingQueue是滿的,任何試圖往里存東西的操作也會被阻斷進(jìn)入等待狀態(tài),直到BlockingQueue里有空間才會被喚醒繼續(xù)操作。 阻塞隊列常用于生產(chǎn)者和消費者的場景,生產(chǎn)者是往隊列里添加元素的線程,消費者是從隊列里拿元素的線程。阻塞隊列就是生產(chǎn)者存放元素的容器,而消費者也只從容器里拿元素。具體的實現(xiàn)類有LinkedBlockingQueue,ArrayBlockingQueued等。一般其內(nèi)部的都是通過Lock和Condition(顯示鎖(Lock)及Condition的學(xué)習(xí)與使用)來實現(xiàn)阻塞和喚醒。
線程池的工作過程如下:
線程池剛創(chuàng)建時,里面沒有一個線程。任務(wù)隊列是作為參數(shù)傳進(jìn)來的。不過,就算隊列里面有任務(wù),線程池也不會馬上執(zhí)行它們。
當(dāng)調(diào)用 execute() 方法添加一個任務(wù)時,線程池會做如下判斷:
如果正在運行的線程數(shù)量小于 corePoolSize,那么馬上創(chuàng)建線程運行這個任務(wù);
如果正在運行的線程數(shù)量大于或等于 corePoolSize,那么將這個任務(wù)放入隊列;
如果這時候隊列滿了,而且正在運行的線程數(shù)量小于 maximumPoolSize,那么還是要創(chuàng)建非核心線程立刻運行這個任務(wù);
如果隊列滿了,而且正在運行的線程數(shù)量大于或等于 maximumPoolSize,那么線程池會拋出異常RejectExecutionException。
當(dāng)一個線程完成任務(wù)時,它會從隊列中取下一個任務(wù)來執(zhí)行。
當(dāng)一個線程無事可做,超過一定的時間(keepAliveTime)時,線程池會判斷,如果當(dāng)前運行的線程數(shù)大于 corePoolSize,那么這個線程就被停掉。所以線程池的所有任務(wù)完成后,它最終會收縮到 corePoolSize 的大小。
線程池的創(chuàng)建和使用
生成線程池采用了工具類Executors的靜態(tài)方法,以下是幾種常見的線程池。
SingleThreadExecutor:單個后臺線程 (其緩沖隊列是無界的)
1
2
3
4
5
6
|
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService ( new ThreadPoolExecutor( 1 , 1 , 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); } |
創(chuàng)建一個單線程的線程池。這個線程池只有一個核心線程在工作,也就是相當(dāng)于單線程串行執(zhí)行所有任務(wù)。如果這個唯一的線程因為異常結(jié)束,那么會有一個新的線程來替代它。此線程池保證所有任務(wù)的執(zhí)行順序按照任務(wù)的提交順序執(zhí)行。
FixedThreadPool:只有核心線程的線程池,大小固定 (其緩沖隊列是無界的) 。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
創(chuàng)建固定大小的線程池。每次提交一個任務(wù)就創(chuàng)建一個線程,直到線程達(dá)到線程池的最大大小。線程池的大小一旦達(dá)到最大值就會保持不變,如果某個線程因為執(zhí)行異常而結(jié)束,那么線程池會補充一個新線程。
CachedThreadPool:無界線程池,可以進(jìn)行自動線程回收。
1
2
3
4
5
|
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor( 0 ,Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); } |
如果線程池的大小超過了處理任務(wù)所需要的線程,那么就會回收部分空閑(60秒不執(zhí)行任務(wù))的線程,當(dāng)任務(wù)數(shù)增加時,此線程池又可以智能的添加新線程來處理任務(wù)。此線程池不會對線程池大小做限制,線程池大小完全依賴于操作系統(tǒng)(或者說JVM)能夠創(chuàng)建的最大線程大小。SynchronousQueue是一個是緩沖區(qū)為1的阻塞隊列。
ScheduledThreadPool:核心線程池固定,大小無限的線程池。此線程池支持定時以及周期性執(zhí)行任務(wù)的需求。
1
2
3
4
5
6
|
public static ExecutorService newScheduledThreadPool( int corePoolSize) { return new ScheduledThreadPool(corePoolSize, Integer.MAX_VALUE, DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS, new DelayedWorkQueue()); } |
創(chuàng)建一個周期性執(zhí)行任務(wù)的線程池。如果閑置,非核心線程池會在DEFAULT_KEEPALIVEMILLIS時間內(nèi)回收。
線程池最常用的提交任務(wù)的方法有兩種:
execute:
ExecutorService.execute(Runnable runable);
submit:
FutureTask task = ExecutorService.submit(Runnable runnable);
FutureTask<T> task = ExecutorService.submit(Runnable runnable,T Result);
FutureTask<T> task = ExecutorService.submit(Callable<T> callable);
submit(Callable callable)的實現(xiàn),submit(Runnable runnable)同理。
1
2
3
4
5
6
|
public <T> Future<T> submit(Callable<T> task) { if (task == null ) throw new NullPointerException(); FutureTask<T> ftask = newTaskFor(task); execute(ftask); return ftask; } |
可以看出submit開啟的是有返回結(jié)果的任務(wù),會返回一個FutureTask對象,這樣就能通過get()方法得到結(jié)果。submit最終調(diào)用的也是execute(Runnable runable),submit只是將Callable對象或Runnable封裝成一個FutureTask對象,因為FutureTask是個Runnable,所以可以在execute中執(zhí)行。關(guān)于Callable對象和Runnable怎么封裝成FutureTask對象,見Callable和Future、FutureTask的使用。
線程池實現(xiàn)的原理
如果只講線程池的使用,那這篇博客沒有什么大的價值,充其量也就是熟悉Executor相關(guān)API的過程。線程池的實現(xiàn)過程沒有用到Synchronized關(guān)鍵字,用的都是Volatile,Lock和同步(阻塞)隊列,Atomic相關(guān)類,F(xiàn)utureTask等等,因為后者的性能更優(yōu)。理解的過程可以很好的學(xué)習(xí)源碼中并發(fā)控制的思想。
在開篇提到過線程池的優(yōu)點是可總結(jié)為以下三點:
線程復(fù)用
控制最大并發(fā)數(shù)
管理線程
1.線程復(fù)用過程
理解線程復(fù)用原理首先應(yīng)了解線程生命周期。
在線程的生命周期中,它要經(jīng)過新建(New)、就緒(Runnable)、運行(Running)、阻塞(Blocked)和死亡(Dead)5種狀態(tài)。
Thread通過new來新建一個線程,這個過程是是初始化一些線程信息,如線程名,id,線程所屬group等,可以認(rèn)為只是個普通的對象。調(diào)用Thread的start()后Java虛擬機會為其創(chuàng)建方法調(diào)用棧和程序計數(shù)器,同時將hasBeenStarted為true,之后調(diào)用start方法就會有異常。
處于這個狀態(tài)中的線程并沒有開始運行,只是表示該線程可以運行了。至于該線程何時開始運行,取決于JVM里線程調(diào)度器的調(diào)度。當(dāng)線程獲取cpu后,run()方法會被調(diào)用。不要自己去調(diào)用Thread的run()方法。之后根據(jù)CPU的調(diào)度在就緒——運行——阻塞間切換,直到run()方法結(jié)束或其他方式停止線程,進(jìn)入dead狀態(tài)。
所以實現(xiàn)線程復(fù)用的原理應(yīng)該就是要保持線程處于存活狀態(tài)(就緒,運行或阻塞)。接下來來看下ThreadPoolExecutor是怎么實現(xiàn)線程復(fù)用的。
在ThreadPoolExecutor主要Worker類來控制線程的復(fù)用。看下Worker類簡化后的代碼,這樣方便理解:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
private final class Worker implements Runnable { final Thread thread; Runnable firstTask; Worker(Runnable firstTask) { this .firstTask = firstTask; this .thread = getThreadFactory().newThread( this ); } public void run() { runWorker( this ); } final void runWorker(Worker w) { Runnable task = w.firstTask; w.firstTask = null ; while (task != null || (task = getTask()) != null ){ task.run(); } } |
Worker是一個Runnable,同時擁有一個thread,這個thread就是要開啟的線程,在新建Worker對象時同時新建一個Thread對象,同時將Worker自己作為參數(shù)傳入TThread,這樣當(dāng)Thread的start()方法調(diào)用時,運行的實際上是Worker的run()方法,接著到runWorker()中,有個while循環(huán),一直從getTask()里得到Runnable對象,順序執(zhí)行。getTask()又是怎么得到Runnable對象的呢?
依舊是簡化后的代碼:
1
2
3
4
5
6
7
8
9
|
private Runnable getTask() { if (一些特殊情況) { return null ; } Runnable r = workQueue.take(); return r; } |
這個workQueue就是初始化ThreadPoolExecutor時存放任務(wù)的BlockingQueue隊列,這個隊列里的存放的都是將要執(zhí)行的Runnable任務(wù)。因為BlockingQueue是個阻塞隊列,BlockingQueue.take()得到如果是空,則進(jìn)入等待狀態(tài)直到BlockingQueue有新的對象被加入時喚醒阻塞的線程。所以一般情況Thread的run()方法就不會結(jié)束,而是不斷執(zhí)行從workQueue里的Runnable任務(wù),這就達(dá)到了線程復(fù)用的原理了。
2.控制最大并發(fā)數(shù)
那Runnable是什么時候放入workQueue?Worker又是什么時候創(chuàng)建,Worker里的Thread的又是什么時候調(diào)用start()開啟新線程來執(zhí)行Worker的run()方法的呢?有上面的分析看出Worker里的runWorker()執(zhí)行任務(wù)時是一個接一個,串行進(jìn)行的,那并發(fā)是怎么體現(xiàn)的呢?
很容易想到是在execute(Runnable runnable)時會做上面的一些任務(wù)。看下execute里是怎么做的。
execute:
簡化后的代碼
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
|
public void execute(Runnable command) { if (command == null ) throw new NullPointerException(); int c = ctl.get(); // 當(dāng)前線程數(shù) < corePoolSize if (workerCountOf(c) < corePoolSize) { // 直接啟動新的線程。 if (addWorker(command, true )) return ; c = ctl.get(); } // 活動線程數(shù) >= corePoolSize // runState為RUNNING && 隊列未滿 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 再次檢驗是否為RUNNING狀態(tài) // 非RUNNING狀態(tài) 則從workQueue中移除任務(wù)并拒絕 if (!isRunning(recheck) && remove(command)) reject(command); // 采用線程池指定的策略拒絕任務(wù) // 兩種情況: // 1.非RUNNING狀態(tài)拒絕新的任務(wù) // 2.隊列滿了啟動新的線程失敗(workCount > maximumPoolSize) } else if (!addWorker(command, false )) reject(command); } |
addWorker:
簡化后的代碼
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
private boolean addWorker(Runnable firstTask, boolean core) { int wc = workerCountOf(c); if (wc >= (core ? corePoolSize : maximumPoolSize)) { return false ; } w = new Worker(firstTask); final Thread t = w.thread; t.start(); } |
根據(jù)代碼再來看上面提到的線程池工作過程中的添加任務(wù)的情況:
* 如果正在運行的線程數(shù)量小于 corePoolSize,那么馬上創(chuàng)建線程運行這個任務(wù);
* 如果正在運行的線程數(shù)量大于或等于 corePoolSize,那么將這個任務(wù)放入隊列;
* 如果這時候隊列滿了,而且正在運行的線程數(shù)量小于 maximumPoolSize,那么還是要創(chuàng)建非核心線程立刻運行這個任務(wù);
* 如果隊列滿了,而且正在運行的線程數(shù)量大于或等于 maximumPoolSize,那么線程池會拋出異常RejectExecutionException。
這就是Android的AsyncTask在并行執(zhí)行是在超出最大任務(wù)數(shù)是拋出RejectExecutionException的原因所在,詳見基于最新版本的AsyncTask源碼解讀及AsyncTask的黑暗面
通過addWorker如果成功創(chuàng)建新的線程成功,則通過start()開啟新線程,同時將firstTask作為這個Worker里的run()中執(zhí)行的第一個任務(wù)。
雖然每個Worker的任務(wù)是串行處理,但如果創(chuàng)建了多個Worker,因為共用一個workQueue,所以就會并行處理了。
所以根據(jù)corePoolSize和maximumPoolSize來控制最大并發(fā)數(shù)。大致過程可用下圖表示。
上面的講解和圖來可以很好的理解的這個過程。
如果是做Android開發(fā)的,并且對Handler原理比較熟悉,你可能會覺得這個圖挺熟悉,其中的一些過程和Handler,Looper,Meaasge使用中,很相似。Handler.send(Message)相當(dāng)于execute(Runnuble),Looper中維護(hù)的Meaasge隊列相當(dāng)于BlockingQueue,只不過需要自己通過同步來維護(hù)這個隊列,Looper中的loop()函數(shù)循環(huán)從Meaasge隊列取Meaasge和Worker中的runWork()不斷從BlockingQueue取Runnable是同樣的道理。
3.管理線程
通過線程池可以很好的管理線程的復(fù)用,控制并發(fā)數(shù),以及銷毀等過程,線程的復(fù)用和控制并發(fā)上面已經(jīng)講了,而線程的管理過程已經(jīng)穿插在其中了,也很好理解。
在ThreadPoolExecutor有個ctl的AtomicInteger變量。通過這一個變量保存了兩個內(nèi)容:
所有線程的數(shù)量 每個線程所處的狀態(tài) 其中低29位存線程數(shù),高3位存runState,通過位運算來得到不同的值。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0 )); //得到線程的狀態(tài) private static int runStateOf( int c) { return c & ~CAPACITY; } //得到Worker的的數(shù)量 private static int workerCountOf( int c) { return c & CAPACITY; } // 判斷線程是否在運行 private static boolean isRunning( int c) { return c < SHUTDOWN; } |
這里主要通過shutdown和shutdownNow()來分析線程池的關(guān)閉過程。首先線程池有五種狀態(tài)來控制任務(wù)添加與執(zhí)行。主要介紹以下三種:
RUNNING狀態(tài):線程池正常運行,可以接受新的任務(wù)并處理隊列中的任務(wù);
SHUTDOWN狀態(tài):不再接受新的任務(wù),但是會執(zhí)行隊列中的任務(wù);
STOP狀態(tài):不再接受新任務(wù),不處理隊列中的任務(wù) shutdown這個方法會將runState置為SHUTDOWN,會終止所有空閑的線程,而仍在工作的線程不受影響,所以隊列中的任務(wù)人會被執(zhí)行。
shutdownNow方法將runState置為STOP。和shutdown方法的區(qū)別,這個方法會終止所有的線程,所以隊列中的任務(wù)也不會被執(zhí)行了。
總結(jié)
通過對ThreadPoolExecutor源碼的分析,從總體上了解了線程池的創(chuàng)建,任務(wù)的添加,執(zhí)行等過程,熟悉這些過程,使用線程池就會更輕松了。
而從中學(xué)到的一些對并發(fā)控制,以及生產(chǎn)者——消費者模型任務(wù)處理的使用,對以后理解或解決其他相關(guān)問題會有很大的幫助。比如Android中的Handler機制,而Looper中的Messager隊列用一個BlookQueue來處理同樣是可以的,這寫就是讀源碼的收獲吧。
以上就是對Java 線程池的資料整理,后續(xù)繼續(xù)補充相關(guān)資料,謝謝大家對本站的支持!