在前面的文章中,我們使用線程的時(shí)候就去創(chuàng)建一個(gè)線程窝革,這樣實(shí)現(xiàn)起來(lái)非常簡(jiǎn)便购城,但是就會(huì)有一個(gè)問(wèn)題:
如果并發(fā)線程數(shù)量很多,并且每個(gè)現(xiàn)場(chǎng)都是執(zhí)行一個(gè)時(shí)間很短的任務(wù)就結(jié)束了虐译,這樣頻繁創(chuàng)建線程就會(huì)大大降低系統(tǒng)的效率工猜,因?yàn)轭l繁創(chuàng)建線程和銷毀線程需要時(shí)間。
那么有沒(méi)有一種八法使得線程可以復(fù)用菱蔬,就是執(zhí)行完一個(gè)任務(wù)篷帅,并不被銷毀,而是可以繼續(xù)執(zhí)其他的任務(wù)拴泌?
在java中可以通過(guò)線程池來(lái)達(dá)到這樣的效果魏身。今天我們就來(lái)詳解一下java的線程池,首先我們從最核心的ThreadPoolExecutor類中的方法講起蚪腐,然后再講訴它的實(shí)現(xiàn)原理箭昵,接著給出了它的使用實(shí)例,最后討論一下如何配置線程池的大小回季。
一家制、Java中的ThreadPoolExecutor類
java.uitl.concurrent.ThreadPoolExecutor類是線程池中最核心的一個(gè)類,因此如果要透徹地了解Java中的線程池泡一,必須先了解這個(gè)類颤殴。下面我們來(lái)看一下ThreadPoolExecutor類的具體實(shí)現(xiàn)源碼。
在ThreadPoolExecutor類中提供了四個(gè)構(gòu)造方法:
public class ThreadPoolExecutor extends AbstractExecutorService {
.....
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
...
}
從上面的代碼可以得知鼻忠,ThreadPoolExecutor繼承了AbstractExecutorService類涵但,并提供了四個(gè)構(gòu)造器,事實(shí)上帖蔓,通過(guò)觀察每個(gè)構(gòu)造器的源碼具體實(shí)現(xiàn)矮瘟,發(fā)現(xiàn)前面三個(gè)構(gòu)造器都是調(diào)用的第四個(gè)構(gòu)造器進(jìn)行的初始化工作。
下面解釋一下構(gòu)造器中各個(gè)參數(shù)的含義:
corePoolSize:核心池的大小塑娇,這個(gè)參數(shù)根后面講訴的線程池的實(shí)現(xiàn)原理有非常大的關(guān)系澈侠。在創(chuàng)建線程池后,默認(rèn)情況下埋酬,線程池并沒(méi)有任何線程哨啃,而是等待任務(wù)到來(lái)才創(chuàng)建線程去執(zhí)行任務(wù)烧栋,除非調(diào)用了prestartAllCoreThreads()或者prestartCoreThread()方法,從這2個(gè)方法的名字就可以看出棘催,是預(yù)創(chuàng)建線程的意思劲弦,即在沒(méi)有任務(wù)到來(lái)之前就創(chuàng)建corePoolSize個(gè)線程或者一個(gè)線程耳标。默認(rèn)情況下醇坝,在創(chuàng)建線程池后,線程池中的線程為0次坡,當(dāng)有任務(wù)的之后呼猪,就會(huì)創(chuàng)建一個(gè)線程去執(zhí)行任務(wù),當(dāng)線程池中的線程數(shù)目達(dá)到corePoolSize后砸琅,就會(huì)把到達(dá)的任務(wù)放到緩存隊(duì)列當(dāng)中宋距;
maxmunPoolSize:線程池最大程數(shù),這個(gè)參數(shù)也是一個(gè)非常重要的參數(shù)症脂,它表示在線程池中最多能創(chuàng)建多少個(gè)線程谚赎;
keepAliveTime:表示線程沒(méi)有任務(wù)執(zhí)行時(shí)最多保持多久時(shí)間會(huì)終止。默認(rèn)情況下诱篷,只有當(dāng)線程池中的線程數(shù)大于corelPoolSize時(shí)壶唤,如果一個(gè)線程空閑的時(shí)間達(dá)到keepAliveTime,則會(huì)終止,直到線程池中的線程數(shù)不超過(guò)corePoolSize棕所。但是如果調(diào)用了allowCoreThreadTimeOut(boolean)方法闸盔,在線程中的線程數(shù)不大于corePoolSize時(shí),keepAliveTime參數(shù)也會(huì)起作用琳省,直到線程池中的線程數(shù)為0迎吵;
unit:參數(shù)keepAliveTime的時(shí)間單位,有7種取值针贬,在TimeUnit類中7種靜態(tài)屬性:
TimeUnit.DAYS; //天
TimeUnit.HOURS; //小時(shí)
TimeUnit.MINUTES; //分鐘
TimeUnit.SECONDS; //秒
TimeUnit.MILLISECONDS; //毫秒
TimeUnit.MICROSECONDS; //微妙
TimeUnit.NANOSECONDS; //納秒
workQueue:一個(gè)阻塞隊(duì)列击费,用來(lái)存儲(chǔ)等待執(zhí)行的任務(wù),這個(gè)參數(shù)的選擇也很重要桦他,會(huì)對(duì)線程池的運(yùn)行過(guò)程產(chǎn)生重大影響荡灾,一般來(lái)說(shuō)這里的阻塞隊(duì)列有以下幾種選擇:
ArrayBlockingQueue;
LinkedBlockingQueue;
SynchronousQueue;
ArrayBlockingQueue和PriorityBlockingQueue使用較少,一般使用LinkedBlockingQueue和Synchronous瞬铸。線程池的排隊(duì)策略與BlockingQueue有關(guān)批幌。
threadFactory:線程工廠,主要用來(lái)創(chuàng)建線程嗓节;
handler:表示拒絕處理任務(wù)時(shí)的策略荧缘,有以下四種取值:
ThreadPoolExecutor.AbortPolicy:丟棄任務(wù)并拋出 RejectedExecutionException異常。
ThreadPoolExecutor.DiscardPolicy:也是丟棄任務(wù)拦宣,但是不拋出異常截粗。
ThreadPoolExecutor.DiscardOldestPolicy:丟棄隊(duì)列最前面的任務(wù)信姓,然后重新嘗試執(zhí)行任務(wù)(重復(fù)此過(guò)程)
ThreadPoolExecutor.CallerRunsPolicy:由調(diào)用線程處理該任務(wù)
具體參數(shù)的配置與線程池的關(guān)系將在下一節(jié)講述。
從上面給出的ThreadPoolExecutor類的代碼可以知道绸罗,ThreadPoolExecutor繼承了AbstractExecutorService意推,我們來(lái)看一下AbstractExecutorService的實(shí)現(xiàn):
public abstract class AbstractExecutorService implements ExecutorService {
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };
public Future<?> submit(Runnable task) {};
public <T> Future<T> submit(Runnable task, T result) { };
public <T> Future<T> submit(Callable<T> task) { };
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
};
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
};
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
};
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
};
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
};
AbstractExecutorService是一個(gè)抽象類,它實(shí)現(xiàn)了ExecutorService接口珊蟀。
我們接著看ExecutorService接口的實(shí)現(xiàn):
public interface ExecutorService extends Executor {
void shutdown();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
而ExecutorService又是繼承了Executor接口菊值,我們看一下Executor接口的實(shí)現(xiàn):
public interface Executor {
void execute(Runnable command);
}
到這里,大家應(yīng)該明白了ThreadPoolExecutor育灸、AbstractExecutorService腻窒、ExecutorService和Executor幾個(gè)之間的關(guān)系了。
Executor是一個(gè)頂層接口磅崭,在它里面只聲明了一個(gè)方法execute(Runnable)儿子,返回值為void,參數(shù)為Runnable類型砸喻,從字面意思可以理解柔逼,就是用來(lái)執(zhí)行傳進(jìn)去的任務(wù)的;
然后ExecutorService接口繼承了Executor接口割岛,并聲明了一些方法:submit愉适、invokeAll、invokeAny以及shutDown等蜂桶;
抽象類AbstractExecutorService實(shí)現(xiàn)了ExecutorService接口儡毕,基本實(shí)現(xiàn)了ExecutorService中聲明的所有方法;
然后ThreadPoolExecutor繼承了類AbstractExecutorService扑媚。
在ThreadPoolExecutor類中有幾個(gè)非常重要的方法:
execute()
submit()
shutdown()
shutdownNow()
execute()方法實(shí)際上是Executor中聲明的方法腰湾,在ThreadPoolExecutor進(jìn)行了具體的實(shí)現(xiàn),這個(gè)方法是ThreadPoolExecutor的核心方法疆股,通過(guò)這個(gè)方法可以向線程池提交一個(gè)任務(wù)费坊,交由線程池去執(zhí)行。
submit()方法是在ExecutorService中聲明的方法旬痹,在AbstractExecutorService就已經(jīng)有了具體的實(shí)現(xiàn)附井,在ThreadPoolExecutor中并沒(méi)有對(duì)其進(jìn)行重寫,這個(gè)方法也是用來(lái)向線程池提交任務(wù)的两残,但是它和execute()方法不同永毅,它能夠返回任務(wù)執(zhí)行的結(jié)果,去看submit()方法的實(shí)現(xiàn)人弓,會(huì)發(fā)現(xiàn)它實(shí)際上還是調(diào)用的execute()方法沼死,只不過(guò)它利用了Future來(lái)獲取任務(wù)執(zhí)行結(jié)果(Future相關(guān)內(nèi)容將在下一篇講述)。
shutdown()和shutdownNow()是用來(lái)關(guān)閉線程池的崔赌。
還有很多其他的方法:
比如:getQueue() 意蛀、getPoolSize() 耸别、getActiveCount()、getCompletedTaskCount()等獲取與線程池相關(guān)屬性的方法县钥,有興趣的朋友可以自行查閱API秀姐。
二、深入解析線程池實(shí)現(xiàn)原理
在上一節(jié)我們從宏觀上介紹了ThreadPoolExecutor若贮,下面我們來(lái)深入解析一下線程池的具體實(shí)現(xiàn)原理省有,將從下面幾個(gè)方面講解:
- 1.線程池的狀態(tài)
- 2.任務(wù)的執(zhí)行
- 3.線程池的線程初始化
- 4.任務(wù)緩存隊(duì)列及排隊(duì)策略
- 5.任務(wù)拒絕策略
- 6.線程池的關(guān)閉
- 7.線程池容量的動(dòng)態(tài)調(diào)整
1.線程池狀態(tài)
在ThreadPoolExecutor中定義了一個(gè)volatile變量,另外定義了幾個(gè)static final變量表示線程池的各個(gè)狀態(tài):
volatile int runState;
static final int RUNNING = 0;
static final int SHUTDOWN = 1;
static final int STOP = 2;
static final int TERMINATED = 3;
runState表示當(dāng)前線程池的狀態(tài)兜看,它是一個(gè)volatile變量用來(lái)保證線程之間的可見(jiàn)性锥咸;
下面的幾個(gè)static final變量表示runState可能的幾個(gè)取值狭瞎。
當(dāng)創(chuàng)建線程池后细移,初始時(shí),線程池處于RUNNING狀態(tài)熊锭;
如果調(diào)用了shutdown()方法弧轧,則線程池處于SHUTDOWN狀態(tài),此時(shí)線程池不能夠接受新的任務(wù)碗殷,它會(huì)等待所有任務(wù)執(zhí)行完畢精绎;
如果調(diào)用了shutdownNow()方法,則線程池處于STOP狀態(tài)锌妻,此時(shí)線程池不能接受新的任務(wù)代乃,并且會(huì)去嘗試終止正在執(zhí)行的任務(wù);
當(dāng)線程池處于SHUTDOWN或STOP狀態(tài)仿粹,并且所有工作線程已經(jīng)銷毀搁吓,任務(wù)緩存隊(duì)列已經(jīng)清空或執(zhí)行結(jié)束后,線程池被設(shè)置為TERMINATED狀態(tài)吭历。
2.任務(wù)的執(zhí)行
在了解將任務(wù)提交給線程池到任務(wù)執(zhí)行完畢整個(gè)過(guò)程之前堕仔,我們先來(lái)看一下ThreadPoolExecutor類中其他的一些比較成員變量:
private final BlockingQueue<Runnable> workQueue; //任務(wù)緩存隊(duì)列,用來(lái)存放等待執(zhí)行的任務(wù)
private final ReentrantLock mainLock = new ReentrantLock(); //線程池的主要狀態(tài)鎖晌区,對(duì)線程池狀態(tài)(比如線程池大小
//摩骨、runState等)的改變都要使用這個(gè)鎖
private final HashSet<Worker> workers = new HashSet<Worker>(); //用來(lái)存放工作集
private volatile long keepAliveTime; //線程存貨時(shí)間
private volatile boolean allowCoreThreadTimeOut; //是否允許為核心線程設(shè)置存活時(shí)間
private volatile int corePoolSize; //核心池的大小(即線程池中的線程數(shù)目大于這個(gè)參數(shù)時(shí)朗若,提交的任務(wù)會(huì)被放進(jìn)任務(wù)緩存隊(duì)列)
private volatile int maximumPoolSize; //線程池最大能容忍的線程數(shù)
private volatile int poolSize; //線程池中當(dāng)前的線程數(shù)
private volatile RejectedExecutionHandler handler; //任務(wù)拒絕策略
private volatile ThreadFactory threadFactory; //線程工廠恼五,用來(lái)創(chuàng)建線程
private int largestPoolSize; //用來(lái)記錄線程池中曾經(jīng)出現(xiàn)過(guò)的最大線程數(shù)
private long completedTaskCount; //用來(lái)記錄已經(jīng)執(zhí)行完畢的任務(wù)個(gè)數(shù)
每個(gè)變量的作用都已經(jīng)表明出來(lái)了,這里要重點(diǎn)解釋下corePoolSize哭懈、maxmumPoolSize灾馒、largestPoolSize三個(gè)變量。
corePoolSize在很多地方被翻譯成核心池大小银伟,其實(shí)我的理解這個(gè)就是線程池的大小你虹。舉個(gè)簡(jiǎn)單的例子:
假如有一個(gè)工廠绘搞,工廠里面有10個(gè)工人,每個(gè)工人同時(shí)只能做一件任務(wù)傅物。
因此只要當(dāng)10個(gè)工人中有工人是空閑的夯辖,來(lái)了任務(wù)就分配給空閑的工人做;
當(dāng)10個(gè)工人都有任務(wù)在做時(shí)董饰,如果還來(lái)了任務(wù)蒿褂,就把任務(wù)進(jìn)行排隊(duì)等待;
如果說(shuō)新任務(wù)數(shù)目增長(zhǎng)的速度遠(yuǎn)遠(yuǎn)大于工人做任務(wù)的速度卒暂,那么此時(shí)工廠主管可能會(huì)想補(bǔ)救措施啄栓,比如重新招4個(gè)臨時(shí)工人進(jìn)來(lái);
然后就將任務(wù)也分配給這4個(gè)臨時(shí)工人做也祠;
如果說(shuō)著14個(gè)工人做任務(wù)的速度還是不夠昙楚,此時(shí)工廠主管可能就要考慮不再接收新的任務(wù)或者拋棄前面的一些任務(wù)了。
當(dāng)這14個(gè)工人當(dāng)中有人空閑時(shí)诈嘿,而新任務(wù)增長(zhǎng)的速度又比較緩慢堪旧,工廠主管可能就考慮辭掉4個(gè)臨時(shí)工了,只保持原來(lái)的10個(gè)工人奖亚,畢竟請(qǐng)額外的工人是要花錢的淳梦。
這個(gè)例子中的corePoolSize就是10,而maximumPoolSize就是14(10+4)昔字。
也就是說(shuō)corePoolSize就是線程池大小,maxmumPoolSize在我看來(lái)線程池的一種補(bǔ)救措施爆袍,即任務(wù)量突然過(guò)大時(shí)一種補(bǔ)救措施。
不過(guò)為了方便理解作郭,在本文后面還是將corePoolSize翻譯成核心池大小陨囊。
largestPoolSize只是一個(gè)用來(lái)起記錄作用的變量,用來(lái)記錄線程池中曾經(jīng)有過(guò)的最大線程數(shù)目所坯,跟線程池的容量沒(méi)有任何關(guān)系谆扎。
下面我們進(jìn)入正題,看一下任務(wù)從提交到最終執(zhí)行完畢經(jīng)歷了那些過(guò)程芹助。
在ThreadPoolExecutor類中堂湖,最核心的任務(wù)提交方法executor()方法,雖然通過(guò)submit也可以提交状土,但是實(shí)際上submit方法里面最終調(diào)用的還是execute()方法无蜂,所以我們只需要研究executor()方法的實(shí)現(xiàn)原理即可:
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
上面的代碼可能看起來(lái)不是那么容易理解,下面我們一句一句解釋:
首先蒙谓,判斷提交的任務(wù)command是否為null斥季,若是null,則拋出空指針異常;
接著是這句酣倾,這句要好好理解一下:
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))
由于是或條件運(yùn)算符舵揭,所以先計(jì)算前半部分的值,如果線程池中當(dāng)前線程數(shù)不小于核心池大小躁锡,那么就會(huì)直接進(jìn)入下面的if語(yǔ)句塊了午绳。
如果線程池中當(dāng)前線程數(shù)小于核心池大小,則接著執(zhí)行后半部分映之,也就是執(zhí)行拦焚。
addIfUnderCorePoolSize(command)
如果執(zhí)行完addIfUnderCorePoolSize這個(gè)方法返回false,則繼續(xù)執(zhí)行下面的if語(yǔ)句塊杠输,否則整個(gè)方法就直接執(zhí)行完畢了赎败。
如果執(zhí)行完addIfUnderCorePoolSize這個(gè)方法返回false,然后接著判斷:
if (runState == RUNNING && workQueue.offer(command))
如果當(dāng)前線程池處于RUNNING狀態(tài)蠢甲,則將任務(wù)放入任務(wù)緩存隊(duì)列僵刮;如果當(dāng)前線程池不處于RUNNING狀態(tài)或者任務(wù)放入緩存隊(duì)列失敗,則執(zhí)行:
addIfUnderMaximumPoolSize(command)
如果執(zhí)行addIfUnderMaximumPoolSize方法失敗峡钓,則執(zhí)行reject()方法進(jìn)行任務(wù)拒絕處理妓笙。
回到前面:
if (runState == RUNNING && workQueue.offer(command))
這句的執(zhí)行若河,如果說(shuō)當(dāng)前線程池處于RUNNING狀態(tài)且將任務(wù)放入任務(wù)緩存隊(duì)列成功能岩,則繼續(xù)進(jìn)行判斷:
if (runState != RUNNING || poolSize == 0)
這句判斷是為了防止在將此任務(wù)添加進(jìn)任務(wù)緩存隊(duì)列的同時(shí)其他線程突然調(diào)用shutdown或者shutdownNow方法關(guān)閉了線程池的一種應(yīng)急措施。如果是這樣就執(zhí)行:
ensureQueuedTaskHandled(command)
進(jìn)行應(yīng)急處理萧福,從名字可以看出是保證 添加到任務(wù)緩存隊(duì)列中的任務(wù)得到處理拉鹃。
我們接著看2個(gè)關(guān)鍵方法的實(shí)現(xiàn):addIfUnderCorePoolSize和addIfUnderMaximumPoolSize:
private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask); //創(chuàng)建線程去執(zhí)行firstTask任務(wù)
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
這個(gè)是addIfUnderCorePoolSize方法的具體實(shí)現(xiàn),從名字可以看出它的意圖就是當(dāng)?shù)陀诤诵某源笮r(shí)執(zhí)行的方法鲫忍。下面看其具體實(shí)現(xiàn)膏燕,首先獲取到鎖,因?yàn)檫@地方涉及到線程池狀態(tài)的變化悟民,先通過(guò)if語(yǔ)句判斷當(dāng)前線程池中的線程數(shù)目是否小于核心池大小坝辫,有朋友也許會(huì)有疑問(wèn):前面在execute()方法中不是已經(jīng)判斷過(guò)了嗎,只有線程池當(dāng)前線程數(shù)目小于核心池大小才會(huì)執(zhí)行addIfUnderCorePoolSize方法的射亏,為何這地方還要繼續(xù)判斷近忙?原因很簡(jiǎn)單,前面的判斷過(guò)程中并沒(méi)有加鎖智润,因此可能在execute方法判斷的時(shí)候poolSize小于corePoolSize及舍,而判斷完之后,在其他線程中又向線程池提交了任務(wù)窟绷,就可能導(dǎo)致poolSize不小于corePoolSize了锯玛,所以需要在這個(gè)地方繼續(xù)判斷。然后接著判斷線程池的狀態(tài)是否為RUNNING兼蜈,原因也很簡(jiǎn)單攘残,因?yàn)橛锌赡茉谄渌€程中調(diào)用了shutdown或者shutdownNow方法拙友。然后就是執(zhí)行
t = addThread(firstTask);
這個(gè)方法也非常關(guān)鍵,傳進(jìn)去的參數(shù)為提交的任務(wù)歼郭,返回值為Thread類型献宫。然后接著在下面判斷t是否為空,為空則表明創(chuàng)建線程失斒等觥(即poolSize>=corePoolSize或者runState不等于RUNNING)姊途,否則調(diào)用t.start()方法啟動(dòng)線程。
我們來(lái)看一下addThread方法的實(shí)現(xiàn):
private Thread addThread(Runnable firstTask) {
Worker w = new Worker(firstTask);
Thread t = threadFactory.newThread(w); //創(chuàng)建一個(gè)線程知态,執(zhí)行任務(wù)
if (t != null) {
w.thread = t; //將創(chuàng)建的線程的引用賦值為w的成員變量
workers.add(w);
int nt = ++poolSize; //當(dāng)前線程數(shù)加1
if (nt > largestPoolSize)
largestPoolSize = nt;
}
return t;
}
在addThread方法中捷兰,首先用提交的任務(wù)創(chuàng)建了一個(gè)Worker對(duì)象,然后調(diào)用線程工廠threadFactory創(chuàng)建了一個(gè)新的線程t负敏,然后將線程t的引用賦值給了Worker對(duì)象的成員變量thread贡茅,接著通過(guò)workers.add(w)將Worker對(duì)象添加到工作集當(dāng)中。
下面我們看一下Worker類的實(shí)現(xiàn):
private final class Worker implements Runnable {
private final ReentrantLock runLock = new ReentrantLock();
private Runnable firstTask;
volatile long completedTasks;
Thread thread;
Worker(Runnable firstTask) {
this.firstTask = firstTask;
}
boolean isActive() {
return runLock.isLocked();
}
void interruptIfIdle() {
final ReentrantLock runLock = this.runLock;
if (runLock.tryLock()) {
try {
if (thread != Thread.currentThread())
thread.interrupt();
} finally {
runLock.unlock();
}
}
}
void interruptNow() {
thread.interrupt();
}
private void runTask(Runnable task) {
final ReentrantLock runLock = this.runLock;
runLock.lock();
try {
if (runState < STOP &&
Thread.interrupted() &&
runState >= STOP)
boolean ran = false;
beforeExecute(thread, task); //beforeExecute方法是ThreadPoolExecutor類的一個(gè)方法其做,沒(méi)有具體實(shí)現(xiàn)顶考,用戶可以根據(jù)
//自己需要重載這個(gè)方法和后面的afterExecute方法來(lái)進(jìn)行一些統(tǒng)計(jì)信息,比如某個(gè)任務(wù)的執(zhí)行時(shí)間等
try {
task.run();
ran = true;
afterExecute(task, null);
++completedTasks;
} catch (RuntimeException ex) {
if (!ran)
afterExecute(task, ex);
throw ex;
}
} finally {
runLock.unlock();
}
}
public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this); //當(dāng)任務(wù)隊(duì)列中沒(méi)有任務(wù)時(shí)妖泄,進(jìn)行清理工作
}
}
它實(shí)際上實(shí)現(xiàn)了Runnable接口驹沿,因此上面的Thread t = threadFactory.newThread(w);效果跟下面這句的效果基本一樣:
Thread t = new Thread(w);
相當(dāng)于傳進(jìn)去了一個(gè)Runnable任務(wù),在線程t中執(zhí)行這個(gè)Runnable蹈胡。
既然Worker實(shí)現(xiàn)了Runnable接口渊季,那么自然最核心的方法便是run()方法了:
public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
從run方法的實(shí)現(xiàn)可以看出,它首先執(zhí)行的是通過(guò)構(gòu)造器傳進(jìn)來(lái)的任務(wù)firstTask罚渐,在調(diào)用runTask()執(zhí)行完firstTask之后却汉,在while循環(huán)里面不斷通過(guò)getTask()去取新的任務(wù)來(lái)執(zhí)行,那么去哪里取呢荷并?自然是從任務(wù)緩存隊(duì)列里面去取合砂,getTask是ThreadPoolExecutor類中的方法,并不是Worker類中的方法源织,下面是getTask方法的實(shí)現(xiàn):
Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN) // Help drain queue
r = workQueue.poll();
else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果線程數(shù)大于核心池大小或者允許為核心池線程設(shè)置空閑時(shí)間翩伪,
//則通過(guò)poll取任務(wù),若等待一定的時(shí)間取不到任務(wù)雀鹃,則返回null
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take();
if (r != null)
return r;
if (workerCanExit()) { //如果沒(méi)取到任務(wù)幻工,即r為null,則判斷當(dāng)前的worker是否可以退出
if (runState >= SHUTDOWN) // Wake up others
interruptIdleWorkers(); //中斷處于空閑狀態(tài)的worker
return null;
}
// Else retry
} catch (InterruptedException ie) {
// On interruption, re-check runState
}
}
在getTask中黎茎,先判斷當(dāng)前線程池狀態(tài)囊颅,如果runState大于SHUTDOWN(即為STOP或者TERMINATED),則直接返回null。
如果runState為SHUTDOWN或者RUNNING踢代,則從任務(wù)緩存隊(duì)列取任務(wù)盲憎。
如果當(dāng)前線程池的線程數(shù)大于核心池大小corePoolSize或者允許為核心池中的線程設(shè)置空閑存活時(shí)間,則調(diào)用poll(time,timeUnit)來(lái)取任務(wù)胳挎,這個(gè)方法會(huì)等待一定的時(shí)間饼疙,如果取不到任務(wù)就返回null。
然后判斷取到的任務(wù)r是否為null慕爬,為null則通過(guò)調(diào)用workerCanExit()方法來(lái)判斷當(dāng)前worker是否可以退出窑眯,我們看一下workerCanExit()的實(shí)現(xiàn):
private boolean workerCanExit() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
boolean canExit;
//如果runState大于等于STOP,或者任務(wù)緩存隊(duì)列為空了
//或者 允許為核心池線程設(shè)置空閑存活時(shí)間并且線程池中的線程數(shù)目大于1
try {
canExit = runState >= STOP ||
workQueue.isEmpty() ||
(allowCoreThreadTimeOut &&
poolSize > Math.max(1, corePoolSize));
} finally {
mainLock.unlock();
}
return canExit;
}
也就是說(shuō)如果線程池處于STOP狀態(tài)医窿,或者任務(wù)隊(duì)列已為空或者允許為核心線程設(shè)置空閑存活時(shí)間并且線程數(shù)大于1時(shí)磅甩,允許worker退出。如果允許worker退出姥卢,則調(diào)用interruptIdleWorkers()中斷處于空閑狀態(tài)的worker卷要,我們看一下interruptIdleWorkers()的實(shí)現(xiàn):
void interruptIdleWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) //實(shí)際上調(diào)用的是worker的interruptIfIdle()方法
w.interruptIfIdle();
} finally {
mainLock.unlock();
}
從實(shí)現(xiàn)可以看出,它實(shí)際上調(diào)用的是worker的interruptIfIdle()方法独榴,在worker的interruptIfIdle()方法中:
void interruptIfIdle() {
final ReentrantLock runLock = this.runLock;
if (runLock.tryLock()) { //注意這里僧叉,是調(diào)用tryLock()來(lái)獲取鎖的,因?yàn)槿绻?dāng)前worker正在執(zhí)行任務(wù)棺榔,鎖已經(jīng)被獲取了瓶堕,是無(wú)法獲取到鎖的
//如果成功獲取了鎖,說(shuō)明當(dāng)前worker處于空閑狀態(tài)
try {
if (thread != Thread.currentThread())
thread.interrupt();
} finally {
runLock.unlock();
}
}
這里有一個(gè)非常巧妙的設(shè)計(jì)方式掷豺,假如我們來(lái)設(shè)計(jì)線程池捞烟,可能會(huì)有一個(gè)任務(wù)分派線程,當(dāng)發(fā)現(xiàn)有線程空閑時(shí)当船,就從任務(wù)緩存隊(duì)列中取一個(gè)任務(wù)交給空閑線程執(zhí)行。但是在這里默辨,并沒(méi)有采用這樣的方式德频,因?yàn)檫@樣會(huì)要額外地對(duì)任務(wù)分派線程進(jìn)行管理,無(wú)形地會(huì)增加難度和復(fù)雜度缩幸,這里直接讓執(zhí)行完任務(wù)的線程去任務(wù)緩存隊(duì)列里面取任務(wù)來(lái)執(zhí)行壹置。
我們?cè)倏碼ddIfUnderMaximumPoolSize方法的實(shí)現(xiàn),這個(gè)方法的實(shí)現(xiàn)思想和addIfUnderCorePoolSize方法的實(shí)現(xiàn)思想非常相似表谊,唯一的區(qū)別在于addIfUnderMaximumPoolSize方法是在線程池中的線程數(shù)達(dá)到了核心池大小并且往任務(wù)隊(duì)列中添加任務(wù)失敗的情況下執(zhí)行的:
private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < maximumPoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
看到?jīng)]有钞护,其實(shí)它和addIfUnderCorePoolSize方法的實(shí)現(xiàn)基本一模一樣,只是if語(yǔ)句判斷條件中的poolSize < maximumPoolSize不同而已爆办。
到這里难咕,大部分朋友應(yīng)該對(duì)任務(wù)提交給線程池之后到被執(zhí)行的整個(gè)過(guò)程有了一個(gè)基本的了解,下面總結(jié)一下:
- 1)首先,要清楚corePoolSize和maximumPoolSize的含義
- 2)其次余佃,要知道Worker是用來(lái)起到什么作用的暮刃;
- 3)要知道任務(wù)提交給線程池之后的處理策略,這里總結(jié)一下主要有4點(diǎn):
- 如果當(dāng)前線程池中的線程數(shù)目小于corePoolSize爆土,則每來(lái)一個(gè)任務(wù)椭懊,就會(huì)創(chuàng)建一個(gè)線程去執(zhí)行這個(gè)任務(wù);
- 如果當(dāng)前線程池中的線程數(shù)目>=corePoolSize,則每來(lái)一個(gè)任務(wù)步势,會(huì)嘗試將其添加到任務(wù)緩存隊(duì)列當(dāng)中氧猬,若添加成功,則該任務(wù)會(huì)等待空閑線程將其取出執(zhí)行坏瘩;若添加失敗(一般來(lái)說(shuō)是任務(wù)緩存隊(duì)列已滿),則會(huì)嘗試創(chuàng)建新的線程去執(zhí)行這個(gè)任務(wù)狂窑;
- 如果當(dāng)前線程池中的線程數(shù)目達(dá)到maxinumPoolSize,則會(huì)采取任務(wù)拒絕策略進(jìn)行處理;
- 如果線程池中的線程數(shù)量大于 corePoolSize時(shí)桑腮,如果某線程空閑時(shí)間超過(guò)keepAliveTime泉哈,線程將被終止,直至線程池中的線程數(shù)目不大于corePoolSize破讨;如果允許為核心池中的線程設(shè)置存活時(shí)間丛晦,那么核心池中的線程空閑時(shí)間超過(guò)keepAliveTime,線程也會(huì)被終止提陶。
線程池中的線程初始化
默認(rèn)情況下烫沙,創(chuàng)建線程池之后,線程池中沒(méi)有線程的隙笆,需要提交任務(wù)之后锌蓄,才會(huì)創(chuàng)建線程。
在實(shí)際中如果需要線程池創(chuàng)建之后立即創(chuàng)建線程撑柔,可以通過(guò)以下兩個(gè)方法辦到:
- prestartCoreThread():初始化一個(gè)核心線程瘸爽;
- prestartAllCoreThreads():初始化所有核心線程
下面是這2個(gè)方法的實(shí)現(xiàn):
public boolean prestartCoreThread() {
return addIfUnderCorePoolSize(null); //注意傳進(jìn)去的參數(shù)是null
}
public int prestartAllCoreThreads() {
int n = 0;
while (addIfUnderCorePoolSize(null))//注意傳進(jìn)去的參數(shù)是null
++n;
return n;
}
注意上面?zhèn)鬟M(jìn)去的參數(shù)是null,根據(jù)第2小節(jié)的分析可知如果傳進(jìn)去的參數(shù)為null铅忿,則最后執(zhí)行線程會(huì)阻塞在getTask方法中的
r = workQueue.take();
即等待任務(wù)隊(duì)列中有任務(wù)剪决。
4.任務(wù)緩存隊(duì)列及排隊(duì)策略
在前面我們多次提到了任務(wù)緩存隊(duì)列,即workQueue檀训,它用來(lái)存放等待執(zhí)行的任務(wù)柑潦。
workQueue的類型為BlockingQueue<Runnable>,通尘欤可以取下面三種類型:
1) ArrayBlockingQueue: 基于數(shù)組的先進(jìn)先出隊(duì)列渗鬼,此隊(duì)列創(chuàng)建時(shí)必須指定大小荧琼;
2)LinkedBlockingQueue:基于鏈表的先進(jìn)先出隊(duì)列譬胎,如果創(chuàng)建時(shí)沒(méi)有指定此隊(duì)列大小差牛,則默認(rèn)為Integer.MAX_VALUE;
3)synchronousQueue:這個(gè)隊(duì)列比較特殊银择,它不會(huì)保存提交的任務(wù)多糠,而是將直接新建一個(gè)線程來(lái)執(zhí)行新來(lái)的任務(wù)。
5.任務(wù)拒絕策略
當(dāng)線程池的任務(wù)緩存隊(duì)列已滿并且線程池中的線程數(shù)目達(dá)到maximumPoolSize浩考,如果還有任務(wù)到來(lái)就會(huì)采取任務(wù)拒絕策略夹孔,通常有以下四種策略:
ThreadPoolExecutor.AbortPolicy:丟棄任務(wù)并拋出
RejectedExecutionException異常。
ThreadPoolExecutor.DiscardPolicy:也是丟棄任務(wù)析孽,但是不拋出異常搭伤。
ThreadPoolExecutor.DiscardOldestPolicy:丟棄隊(duì)列最前面的任務(wù),然后重新嘗試執(zhí)行任務(wù)(重復(fù)此過(guò)程)
ThreadPoolExecutor.CallerRunsPolicy:由調(diào)用線程處理該任務(wù)
6.線程池的關(guān)閉
ThreadPoolExecutor提供了兩個(gè)方法袜瞬,用于線程池的關(guān)閉怜俐,分別是shutdown()和shutdownNow(),其中:
- shutdown():不會(huì)立即終止線程池邓尤,而是要等所有任務(wù)緩存隊(duì)列中的任務(wù)都執(zhí)行完后才終止拍鲤,但再也不會(huì)接受新的任務(wù)
-
shutdownNow():立即終止線程池,并嘗試打斷正在執(zhí)行的任務(wù)汞扎,并且清空任務(wù)緩存隊(duì)列季稳,返回尚未執(zhí)行的任務(wù)
7.線程池容量的動(dòng)態(tài)調(diào)整
ThreadPoolExecutor提供了動(dòng)態(tài)調(diào)整線程池容量大小的方法:setCorePoolSize()和setMaximumPoolSize(), - setCorePoolSize:設(shè)置核心池大小
- setMaximumPoolSize:設(shè)置線程池最大能創(chuàng)建的線程數(shù)目大小
當(dāng)上述參數(shù)從小變大時(shí)澈魄,ThreadPoolExecutor進(jìn)行線程賦值景鼠,還可能立即創(chuàng)建新的線程來(lái)執(zhí)行任務(wù)。
三痹扇、使用實(shí)例
前面我們討論了關(guān)于線程池的實(shí)現(xiàn)原理铛漓,這一節(jié)我們來(lái)看一下它的具體使用:
public class Test {
public static void main(String[] args) {
ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<Runnable>(5));
for(int i=0;i<15;i++){
MyTask myTask = new MyTask(i);
executor.execute(myTask);
System.out.println("線程池中線程數(shù)目:"+executor.getPoolSize()+",隊(duì)列中等待執(zhí)行的任務(wù)數(shù)目:"+
executor.getQueue().size()+"鲫构,已執(zhí)行玩別的任務(wù)數(shù)目:"+executor.getCompletedTaskCount());
}
executor.shutdown();
}
}
class MyTask implements Runnable {
private int taskNum;
public MyTask(int num) {
this.taskNum = num;
}
@Override
public void run() {
System.out.println("正在執(zhí)行task "+taskNum);
try {
Thread.currentThread().sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("task "+taskNum+"執(zhí)行完畢");
}
從執(zhí)行結(jié)果可以看出浓恶,當(dāng)線程池中線程的數(shù)目大于5時(shí),便將任務(wù)放入任務(wù)緩存隊(duì)列里面芬迄,當(dāng)任務(wù)緩存隊(duì)列滿了之后问顷,便創(chuàng)建新的線程。如果上面程序中禀梳,將for循環(huán)中改成執(zhí)行20個(gè)任務(wù),就會(huì)拋出任務(wù)拒絕異常了肠骆。
不過(guò)在java doc中算途,并不提倡我們直接使用ThreadPoolExecutor,而是使用Executors類中提供的幾個(gè)靜態(tài)方法來(lái)創(chuàng)建線程池:
Executors.newCachedThreadPool(); //創(chuàng)建一個(gè)緩沖池蚀腿,緩沖池容量大小為Integer.MAX_VALUE
Executors.newSingleThreadExecutor(); //創(chuàng)建容量為1的緩沖池
Executors.newFixedThreadPool(int); //創(chuàng)建固定容量大小的緩沖池
下面是這三個(gè)靜態(tài)方法的具體實(shí)現(xiàn);
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
從它們的具體實(shí)現(xiàn)來(lái)看嘴瓤,它們實(shí)際上也是調(diào)用了ThreadPoolExecutor扫外,只不過(guò)參數(shù)都已配置好了。
newFixedThreadPool創(chuàng)建的線程池corePoolSize和maximumPoolSize值是相等的廓脆,它使用的LinkedBlockingQueue筛谚;
newSingleThreadExecutor將corePoolSize和maximumPoolSize都設(shè)置為1,也使用的LinkedBlockingQueue停忿;
newCachedThreadPool將corePoolSize設(shè)置為0驾讲,將maximumPoolSize設(shè)置為Integer.MAX_VALUE,使用的SynchronousQueue席赂,也就是說(shuō)來(lái)了任務(wù)就創(chuàng)建線程運(yùn)行吮铭,當(dāng)線程空閑超過(guò)60秒,就銷毀線程颅停。
實(shí)際中谓晌,如果Executors提供的三個(gè)靜態(tài)方法能滿足要求,就盡量使用它提供的三個(gè)方法癞揉,因?yàn)樽约喝ナ謩?dòng)配置ThreadPoolExecutor的參數(shù)有點(diǎn)麻煩纸肉,要根據(jù)實(shí)際任務(wù)的類型和數(shù)量來(lái)進(jìn)行配置。
另外喊熟,如果ThreadPoolExecutor達(dá)不到要求柏肪,可以自己繼承ThreadPoolExecutor類進(jìn)行重寫。
四逊移、如何合理配置線程池的大小
本節(jié)來(lái)討論一個(gè)比較重要的話題:如何合理配置線程池大小预吆,僅供參考。
一般需要根據(jù)任務(wù)的類型來(lái)配置線程池大懈烊:
如果是CPU密集型任務(wù)拐叉,就需要盡量壓榨CPU,參考值可以設(shè)為 NCPU+1
如果是IO密集型任務(wù)扇商,參考值可以設(shè)置為2*NCPU
當(dāng)然凤瘦,這只是一個(gè)參考值,具體的設(shè)置還需要根據(jù)實(shí)際情況進(jìn)行調(diào)整案铺,比如可以先將線程池大小設(shè)置為參考值蔬芥,再觀察任務(wù)運(yùn)行情況和系統(tǒng)負(fù)載、資源利用率來(lái)進(jìn)行適當(dāng)調(diào)整控汉。