要想理解清楚java線程池實(shí)現(xiàn)原理，明白下面幾個(gè)問(wèn)題就可以了：
專(zhuān)注于為中小企業(yè)提供網(wǎng)站設(shè)計(jì)、成都做網(wǎng)站服務(wù),電腦端+手機(jī)端+微信端的三站合一,更高效的管理,為中小企業(yè)凌河免費(fèi)做網(wǎng)站提供優(yōu)質(zhì)的服務(wù)。我們立足成都，凝聚了一批互聯(lián)網(wǎng)行業(yè)人才，有力地推動(dòng)了上千多家企業(yè)的穩(wěn)健成長(zhǎng)，幫助中小企業(yè)通過(guò)網(wǎng)站建設(shè)實(shí)現(xiàn)規(guī)模擴(kuò)充和轉(zhuǎn)變。
(1)：線程池存在哪些狀態(tài)，這些狀態(tài)之間是如何進(jìn)行切換的呢？
(2)：線程池的種類(lèi)有哪些？
(3)：創(chuàng)建線程池需要哪些參數(shù)，這些參數(shù)的具體含義是什么？
(4)：將任務(wù)添加到線程池之后運(yùn)行流程？
(5)：線程池是怎么做到重用線程的呢？
(6)：線程池的關(guān)閉
首先回答第一個(gè)問(wèn)題：線程池存在哪些狀態(tài)；
查看ThreadPoolExecutor源碼便知曉：
[java]?view plain?copy
//?runState?is?stored?in?the?high-order?bits
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;
存在5種狀態(tài)：
1Running：可以接受新任務(wù)，同時(shí)也可以處理阻塞隊(duì)列里面的任務(wù)；
2Shutdown：不可以接受新任務(wù)，但是可以處理阻塞隊(duì)列里面的任務(wù)；
3Stop：不可以接受新任務(wù)，也不處理阻塞隊(duì)列里面的任務(wù)，同時(shí)還中斷正在處理的任務(wù)；
4Tidying：屬于過(guò)渡階段，在這個(gè)階段表示所有的任務(wù)已經(jīng)執(zhí)行結(jié)束了，當(dāng)前線程池中是不存在有效的線程的，并且將要調(diào)用terminated方法；
5Terminated：終止?fàn)顟B(tài)，這個(gè)狀態(tài)是在調(diào)用完terminated方法之后所處的狀態(tài)；
那么這5種狀態(tài)之間是如何進(jìn)行轉(zhuǎn)換的呢？查看ThreadPoolExecutor源碼里面的注釋便可以知道啦：
[java]?view plain?copy
*?RUNNING?-?SHUTDOWN
*????On?invocation?of?shutdown(),?perhaps?implicitly?in?finalize()
*?(RUNNING?or?SHUTDOWN)?-?STOP
*????On?invocation?of?shutdownNow()
*?SHUTDOWN?-?TIDYING
*????When?both?queue?and?pool?are?empty
*?STOP?-?TIDYING
*????When?pool?is?empty
*?TIDYING?-?TERMINATED
*????When?the?terminated()?hook?method?has?completed
從上面可以看到，在調(diào)用shutdown方法的時(shí)候，線程池狀態(tài)會(huì)從Running轉(zhuǎn)換成Shutdown；在調(diào)用shutdownNow方法的時(shí)候，線程池狀態(tài)會(huì)從Running/Shutdown轉(zhuǎn)換成Stop；在阻塞隊(duì)列為空同時(shí)線程池為空的情況下，線程池狀態(tài)會(huì)從Shutdown轉(zhuǎn)換成Tidying；在線程池為空的情況下，線程池狀態(tài)會(huì)從Stop轉(zhuǎn)換成Tidying；當(dāng)調(diào)用terminated方法之后，線程池狀態(tài)會(huì)從Tidying轉(zhuǎn)換成Terminate；
在明白了線程池的各個(gè)狀態(tài)以及狀態(tài)之間是怎么進(jìn)行切換之后，我們來(lái)看看第二個(gè)問(wèn)題，線程池的種類(lèi)：
(1)：CachedThreadPool：緩存線程池，該類(lèi)線程池中線程的數(shù)量是不確定的，理論上可以達(dá)到Integer.MAX_VALUE個(gè)，這種線程池中的線程都是非核心線程，既然是非核心線程，那么就存在超時(shí)淘汰機(jī)制了，當(dāng)里面的某個(gè)線程空閑時(shí)間超過(guò)了設(shè)定的超時(shí)時(shí)間的話，就會(huì)回收掉該線程；
(2)：FixedThreadPool：固定線程池，這類(lèi)線程池中是只存在核心線程的，對(duì)于核心線程來(lái)說(shuō)，如果我們不設(shè)置allowCoreThreadTimeOut屬性的話是不存在超時(shí)淘汰機(jī)制的，這類(lèi)線程池中的corePoolSize的大小是等于maximumPoolSize大小的，也就是說(shuō)，如果線程池中的線程都處于活動(dòng)狀態(tài)的話，如果有新任務(wù)到來(lái)，他是不會(huì)開(kāi)辟新的工作線程來(lái)處理這些任務(wù)的，只能將這些任務(wù)放到阻塞隊(duì)列里面進(jìn)行等到，直到有核心線程空閑為止；
(3)：ScheduledThreadPool：任務(wù)線程池，這種線程池中核心線程的數(shù)量是固定的，而對(duì)于非核心線程的數(shù)量是不限制的，同時(shí)對(duì)于非核心線程是存在超時(shí)淘汰機(jī)制的，主要適用于執(zhí)行定時(shí)任務(wù)或者周期性任務(wù)的場(chǎng)景；
(4)：SingleThreadPool：?jiǎn)我痪€程池，線程池里面只有一個(gè)線程，同時(shí)也不存在非核心線程，感覺(jué)像是FixedThreadPool的特殊版本，他主要用于確保任務(wù)在同一線程中的順序執(zhí)行，有點(diǎn)類(lèi)似于進(jìn)行同步吧；
接下來(lái)我們來(lái)看第三個(gè)問(wèn)題，創(chuàng)建線程池需要哪些參數(shù)：
同樣查看ThreadPoolExecutor源碼，查看創(chuàng)建線程池的構(gòu)造函數(shù)：
[java]?view plain?copy
public?ThreadPoolExecutor(int?corePoolSize,
int?maximumPoolSize,
long?keepAliveTime,
TimeUnit?unit,
BlockingQueueRunnable?workQueue,
ThreadFactory?threadFactory,
RejectedExecutionHandler?handler)
不管你調(diào)用的是ThreadPoolExecutor的哪個(gè)構(gòu)造函數(shù)，最終都會(huì)執(zhí)行到這個(gè)構(gòu)造函數(shù)的，這個(gè)構(gòu)造函數(shù)有7個(gè)參數(shù)，正是由于對(duì)這7個(gè)參數(shù)值的賦值不同，造成生成不同類(lèi)型的線程池，比如我們常見(jiàn)的CachedThreadPoolExecutor、FixedThreadPoolExecutor
SingleThreadPoolExecutor、ScheduledThreadPoolExecutor，我們老看看這幾個(gè)參數(shù)的具體含義：
1corePoolSize：線程池中核心線程的數(shù)量；當(dāng)提交一個(gè)任務(wù)到線程池的時(shí)候，線程池會(huì)創(chuàng)建一個(gè)線程來(lái)執(zhí)行執(zhí)行任務(wù)，即使有其他空閑的線程存在，直到線程數(shù)達(dá)到corePoolSize時(shí)不再創(chuàng)建，這時(shí)候會(huì)把提交的新任務(wù)放入到阻塞隊(duì)列中，如果調(diào)用了線程池的preStartAllCoreThreads方法，則會(huì)在創(chuàng)建線程池的時(shí)候初始化出來(lái)核心線程；
2maximumPoolSize：線程池允許創(chuàng)建的最大線程數(shù)；如果阻塞隊(duì)列已經(jīng)滿(mǎn)了，同時(shí)已經(jīng)創(chuàng)建的線程數(shù)小于最大線程數(shù)的話，那么會(huì)創(chuàng)建新的線程來(lái)處理阻塞隊(duì)列中的任務(wù)；
3keepAliveTime：線程活動(dòng)保持時(shí)間，指的是工作線程空閑之后繼續(xù)存活的時(shí)間，默認(rèn)情況下，這個(gè)參數(shù)只有線程數(shù)大于corePoolSize的時(shí)候才會(huì)起作用，即當(dāng)線程池中的線程數(shù)目大于corePoolSize的時(shí)候，如果某一個(gè)線程的空閑時(shí)間達(dá)到keepAliveTime，那么這個(gè)線程是會(huì)被終止的，直到線程池中的線程數(shù)目不大于corePoolSize；如果調(diào)用allowCoreThreadTimeOut的話，在線程池中線程數(shù)量不大于corePoolSize的時(shí)候，keepAliveTime參數(shù)也可以起作用的，知道線程數(shù)目為0為止；
4unit：參數(shù)keepAliveTime的時(shí)間單位；
5workQueue：阻塞隊(duì)列；用于存儲(chǔ)等待執(zhí)行的任務(wù)，有四種阻塞隊(duì)列類(lèi)型，ArrayBlockingQueue(基于數(shù)組的有界阻塞隊(duì)列)、LinkedBlockingQueue(基于鏈表結(jié)構(gòu)的阻塞隊(duì)列)、SynchronousQueue(不存儲(chǔ)元素的阻塞隊(duì)列)、PriorityBlockingQueue(具有優(yōu)先級(jí)的阻塞隊(duì)列)；
6threadFactory：用于創(chuàng)建線程的線程工廠；
7handler：當(dāng)阻塞隊(duì)列滿(mǎn)了，且沒(méi)有空閑線程的情況下，也就是說(shuō)這個(gè)時(shí)候，線程池中的線程數(shù)目已經(jīng)達(dá)到了最大線程數(shù)量，處于飽和狀態(tài)，那么必須采取一種策略來(lái)處理新提交的任務(wù)，我們可以自己定義處理策略，也可以使用系統(tǒng)已經(jīng)提供給我們的策略，先來(lái)看看系統(tǒng)為我們提供的4種策略，AbortPolicy(直接拋出異常)、CallerRunsPolicy(只有調(diào)用者所在的線程來(lái)運(yùn)行任務(wù))、DiscardOldestPolicy(丟棄阻塞隊(duì)列中最近的一個(gè)任務(wù)，并執(zhí)行當(dāng)前任務(wù))、Discard(直接丟棄)；
接下來(lái)就是將任務(wù)添加到線程池之后的運(yùn)行流程了；
我們可以調(diào)用submit或者execute方法，兩者最大的區(qū)別在于，調(diào)用submit方法的話，我們可以傳入一個(gè)實(shí)現(xiàn)Callable接口的對(duì)象，進(jìn)而能在當(dāng)前任務(wù)執(zhí)行結(jié)束之后通過(guò)Future對(duì)象獲得任務(wù)的返回值，submit內(nèi)部實(shí)際上還是執(zhí)行的execute方法；而調(diào)用execute方法的話，是不能獲得任務(wù)執(zhí)行結(jié)束之后的返回值的；此外，調(diào)用submit方法的話是可以拋出異常的，但是調(diào)用execute方法的話，異常在其內(nèi)部得到了消化，也就是說(shuō)異常在其內(nèi)部得到了處理，不會(huì)向外傳遞的；
因?yàn)閟ubmit方法最終也是會(huì)執(zhí)行execute方法的，因此我們只需要了解execute方法就可以了：
在execute方法內(nèi)部會(huì)分三種情況來(lái)進(jìn)行處理：
1：首先判斷當(dāng)前線程池中的線程數(shù)量是否小于corePoolSize，如果小于的話，則直接通過(guò)addWorker方法創(chuàng)建一個(gè)新的Worker對(duì)象來(lái)執(zhí)行我們當(dāng)前的任務(wù)；
2：如果說(shuō)當(dāng)前線程池中的線程數(shù)量大于corePoolSize的話，那么會(huì)嘗試將當(dāng)前任務(wù)添加到阻塞隊(duì)列中，然后第二次檢查線程池的狀態(tài)，如果線程池不在Running狀態(tài)的話，會(huì)將剛剛添加到阻塞隊(duì)列中的任務(wù)移出，同時(shí)拒絕當(dāng)前任務(wù)請(qǐng)求；如果第二次檢查發(fā)現(xiàn)當(dāng)前線程池處于Running狀態(tài)的話，那么會(huì)查看當(dāng)前線程池中的工作線程數(shù)量是否為0，如果為0的話，就會(huì)通過(guò)addWorker方法創(chuàng)建一個(gè)Worker對(duì)象出來(lái)處理阻塞隊(duì)列中的任務(wù)；
3：如果原先線程池就不處于Running狀態(tài)或者我們剛剛將當(dāng)前任務(wù)添加到阻塞隊(duì)列的時(shí)候出現(xiàn)錯(cuò)誤的話，那么會(huì)去嘗試通過(guò)addWorker創(chuàng)建新的Worker來(lái)處理當(dāng)前任務(wù)，如果添加失敗的話，則拒絕當(dāng)前任務(wù)請(qǐng)求；
可以看到在上面的execute方法中，我們僅僅只是檢查了當(dāng)前線程池中的線程數(shù)量有沒(méi)有超過(guò)corePoolSize的情況，那么當(dāng)前線程池中的線程數(shù)量有沒(méi)有超過(guò)maximumPoolSize是在哪里檢測(cè)的呢？實(shí)際上是在addWorker方法里面了，我們可以看下addWorker里面的一段代碼：
[java]?view plain?copy
if?(wc?=?CAPACITY?||
wc?=?(core???corePoolSize?:?maximumPoolSize))
return?false;
如果當(dāng)前線程數(shù)量超過(guò)maximumPoolSize的話，直接就會(huì)調(diào)用return方法，返回false；
其實(shí)到這里我們很明顯可以知道，一個(gè)線程池中線程的數(shù)量實(shí)際上就是這個(gè)線程池中Worker的數(shù)量，如果Worker的大小超過(guò)了corePoolSize，那么任務(wù)都在阻塞隊(duì)列里面了，Worker是Java對(duì)我們?nèi)蝿?wù)的一個(gè)封裝類(lèi)，他的聲明是醬紫的：
[java]?view plain?copy
private?final?class?Worker
extends?AbstractQueuedSynchronizer
implements?Runnable
可以看到他實(shí)現(xiàn)了Runnable接口，他是在addWorker方法里面通過(guò)new Worker(firstTask)創(chuàng)建的，我們來(lái)看看他的構(gòu)造函數(shù)就知道了：
[java]?view plain?copy
Worker(Runnable?firstTask)?{
setState(-1);?//?inhibit?interrupts?until?runWorker
this.firstTask?=?firstTask;
this.thread?=?getThreadFactory().newThread(this);
}
而這里的firstTask其實(shí)就是我們調(diào)用execute或者submit的時(shí)候傳入的那個(gè)參數(shù)罷了，一般來(lái)說(shuō)這些參數(shù)是實(shí)現(xiàn)Callable或者Runnable接口的；
在通過(guò)addWorker方法創(chuàng)建出來(lái)Worker對(duì)象之后，這個(gè)方法的最后會(huì)執(zhí)行Worker內(nèi)部thread屬性的start方法，而這個(gè)thread屬性實(shí)際上就是封裝了Worker的Thread，執(zhí)行他的start方法實(shí)際上執(zhí)行的是Worker的run方法，因?yàn)閃orker是實(shí)現(xiàn)了Runnable接口的，在run方法里面就會(huì)執(zhí)行runWorker方法，而runWorker方法里面首先會(huì)判斷當(dāng)前我們傳入的任務(wù)是否為空，不為空的話直接就會(huì)執(zhí)行我們通過(guò)execute或者submit方法提交的任務(wù)啦，注意一點(diǎn)就是我們雖然會(huì)通過(guò)submit方法提交實(shí)現(xiàn)了Callable接口的對(duì)象，但是在調(diào)用submit方法的時(shí)候，其實(shí)是會(huì)將Callable對(duì)象封裝成實(shí)現(xiàn)了Runnable接口對(duì)象的，不信我們看看submit方法源碼是怎么實(shí)現(xiàn)的：
[java]?view plain?copy
public?T?FutureT?submit(CallableT?task)?{
if?(task?==?null)?throw?new?NullPointerException();
RunnableFutureT?ftask?=?newTaskFor(task);
execute(ftask);
return?ftask;
}
看到?jīng)]有呢，實(shí)際上在你傳入實(shí)現(xiàn)了Callable接口對(duì)象的時(shí)候，在submit方法里面是會(huì)將其封裝成RunnableFuture對(duì)象的，而RunnableFuture接口是繼承了Runnable接口的；那么說(shuō)白了其實(shí)就是直接執(zhí)行我們提交任務(wù)的run方法了；如果為空的話，則會(huì)通過(guò)getTask方法從阻塞隊(duì)列里面拿出一個(gè)任務(wù)去執(zhí)行；在任務(wù)執(zhí)行結(jié)束之后繼續(xù)從阻塞隊(duì)列里面拿任務(wù)，直到getTask的返回值為空則退出runWorker內(nèi)部循環(huán)，那么什么情況下getTask返回為空呢？查看getTask方法的源碼注釋可以知道：在Worker必須需要退出的情況下getTask會(huì)返回空，具體什么情況下Worker會(huì)退出呢？(1)：當(dāng)Worker的數(shù)量超過(guò)maximumPoolSize的時(shí)候；(2)：當(dāng)線程池狀態(tài)為Stop的時(shí)候；(3)：當(dāng)線程池狀態(tài)為Shutdown并且阻塞隊(duì)列為空的時(shí)候；(4)：使用等待超時(shí)時(shí)間從阻塞隊(duì)列中拿數(shù)據(jù)，但是超時(shí)之后仍然沒(méi)有拿到數(shù)據(jù)；
如果runWorker方法退出了它里面的循環(huán)，那么就說(shuō)明當(dāng)前阻塞隊(duì)列里面是沒(méi)有任務(wù)可以執(zhí)行的了，你可以看到在runWorker方法內(nèi)部的finally語(yǔ)句塊中執(zhí)行了processWorkerExit方法，用來(lái)對(duì)Worker對(duì)象進(jìn)行回收操作，這個(gè)方法會(huì)傳入一個(gè)參數(shù)表示需要?jiǎng)h除的Worker對(duì)象；在進(jìn)行Worker回收的時(shí)候會(huì)調(diào)用tryTerminate方法來(lái)嘗試關(guān)閉線程池，在tryTerminate方法里面會(huì)檢查是否有Worker在工作，檢查線程池的狀態(tài)，沒(méi)問(wèn)題的話就會(huì)將當(dāng)前線程池的狀態(tài)過(guò)渡到Tidying，之后調(diào)用terminated方法，將線程池狀態(tài)更新到Terminated；
從上面的分析中，我們可以看出線程池運(yùn)行的4個(gè)階段：
(1)：poolSize corePoolSize，則直接創(chuàng)建新的線程(核心線程)來(lái)執(zhí)行當(dāng)前提交的任務(wù)；
(2)：poolSize = corePoolSize，并且此時(shí)阻塞隊(duì)列沒(méi)有滿(mǎn)，那么會(huì)將當(dāng)前任務(wù)添加到阻塞隊(duì)列中，如果此時(shí)存在工作線程(非核心線程)的話，那么會(huì)由工作線程來(lái)處理該阻塞隊(duì)列中的任務(wù)，如果此時(shí)工作線程數(shù)量為0的話，那么會(huì)創(chuàng)建一個(gè)工作線程(非核心線程)出來(lái)；
(3)：poolSize = corePoolSize，并且此時(shí)阻塞隊(duì)列已經(jīng)滿(mǎn)了，那么會(huì)直接創(chuàng)建新的工作線程(非核心線程)來(lái)處理阻塞隊(duì)列中的任務(wù)；
(4)：poolSize = maximumPoolSize，并且此時(shí)阻塞隊(duì)列也滿(mǎn)了的話，那么會(huì)觸發(fā)拒絕機(jī)制，具體決絕策略采用的是什么就要看我們創(chuàng)建ThreadPoolExecutor的時(shí)候傳入的RejectExecutionHandler參數(shù)了；
接下來(lái)就是線程池是怎么做到重用線程的呢？
個(gè)人認(rèn)為線程池里面重用線程的工作是在getTask里面實(shí)現(xiàn)的，在getTask里面是存在兩個(gè)for死循環(huán)嵌套的，他會(huì)不斷的從阻塞對(duì)列里面取出需要執(zhí)行的任務(wù)，返回給我們的runWorker方法里面，而在runWorker方法里面只要getTask返回的任務(wù)不是空就會(huì)執(zhí)行該任務(wù)的run方法來(lái)處理它，這樣一直執(zhí)行下去，直到getTask返回空為止，此時(shí)的情況就是阻塞隊(duì)列里面沒(méi)有任務(wù)了，這樣一個(gè)線程處理完一個(gè)任務(wù)之后接著再處理阻塞隊(duì)列中的另一個(gè)任務(wù)，當(dāng)然在線程池中的不同線程是可以并發(fā)處理阻塞隊(duì)列中的任務(wù)的，最后在阻塞隊(duì)列內(nèi)部不存在任務(wù)的時(shí)候會(huì)去判斷是否需要回收Worker對(duì)象，其實(shí)Worker對(duì)象的個(gè)數(shù)就是線程池中線程的個(gè)數(shù)，至于什么情況才需要回收，上面已經(jīng)說(shuō)了，就是四種情況了；
最后就是線程池是怎樣被關(guān)閉的呢？
涉及到線程池的關(guān)閉，需要用到兩個(gè)方法，shutdown和shutdownNow，他們都是位于ThreadPoolExecutor里面的，對(duì)于shutdown的話，他會(huì)將線程池狀態(tài)切換成Shutdown，此時(shí)是不會(huì)影響對(duì)阻塞隊(duì)列中任務(wù)執(zhí)行的，但是會(huì)拒絕執(zhí)行新加進(jìn)來(lái)的任務(wù)，同時(shí)會(huì)回收閑置的Worker；而shutdownNow方法會(huì)將線程池狀態(tài)切換成Stop，此時(shí)既不會(huì)再去處理阻塞隊(duì)列里面的任務(wù)，也不會(huì)去處理新加進(jìn)來(lái)的任務(wù)，同時(shí)會(huì)回收所有Worker；
線程池通俗的描述就是預(yù)先創(chuàng)建若干空閑線程 等到需要用多線程去處理事務(wù)的時(shí)候去喚醒某些空閑線程執(zhí)行處理任務(wù) 這樣就省去了頻繁創(chuàng)建線程的時(shí)間 因?yàn)轭l 繁創(chuàng)建線程是要耗費(fèi)大量的CPU資源的 如果一個(gè)應(yīng)用程序需要頻繁地處理大量并發(fā)事務(wù) 不斷的創(chuàng)建銷(xiāo)毀線程往往會(huì)大大地降低系統(tǒng)的效率 這時(shí)候線程池就派 上用場(chǎng)了
本文旨在使用Java語(yǔ)言編寫(xiě)一個(gè)通用的線程池 當(dāng)需要使用線程池處理事務(wù)時(shí) 只需按照指定規(guī)范封裝好事務(wù)處理對(duì)象 然后用已有的線程池對(duì)象去自動(dòng)選擇空 閑線程自動(dòng)調(diào)用事務(wù)處理對(duì)象即可 并實(shí)現(xiàn)線程池的動(dòng)態(tài)修改（修改當(dāng)前線程數(shù) 最大線程數(shù)等） 下面是實(shí)現(xiàn)代碼
//ThreadTask java
package polarman threadpool;
/** *//**
*線程任務(wù)
* @author ryang
*
*/
public interface ThreadTask {
public void run();
}
//PooledThread java
package polarman threadpool;
import java util Collection; import java util Vector;
/** *//**
*接受線程池管理的線程
* @author ryang
*
*/
public class PooledThread extends Thread {
protected Vector tasks = new Vector();
protected boolean running = false;
protected boolean stopped = false;
protected boolean paused = false;
protected boolean killed = false;
private ThreadPool pool;
public PooledThread(ThreadPool pool) { this pool = pool;
}
public void putTask(ThreadTask task) { tasks add(task);
}
public void putTasks(ThreadTask[] tasks) { for(int i= ; itasks length; i++) this tasks add(tasks[i]);
}
public void putTasks(Collection tasks) { this tasks addAll(tasks);
}
protected ThreadTask popTask() { if(tasks size() ) return (ThreadTask)tasks remove( );
else
return null;
}
public boolean isRunning() {
return running;
}
public void stopTasks() {
stopped = true;
}
public void stopTasksSync() {
stopTasks();
while(isRunning()) { try {
sleep( );
} catch (InterruptedException e) {
}
}
}
public void pauseTasks() {
paused = true;
}
public void pauseTasksSync() {
pauseTasks();
while(isRunning()) { try {
sleep( );
} catch (InterruptedException e) {
}
}
}
public void kill() { if(!running)
interrupt();
else
killed = true;
}
public void killSync() {
kill();
while(isAlive()) { try {
sleep( );
} catch (InterruptedException e) {
}
}
}
public synchronized void startTasks() {
running = true;
this notify();
}
public synchronized void run() { try { while(true) { if(!running || tasks size() == ) { pool notifyForIdleThread(); //System out println(Thread currentThread() getId() + : 空閑 ); this wait(); }else {
ThreadTask task;
while((task = popTask()) != null) { task run(); if(stopped) {
stopped = false;
if(tasks size() ) { tasks clear(); System out println(Thread currentThread() getId() + : Tasks are stopped );
break;
}
}
if(paused) {
paused = false;
if(tasks size() ) { System out println(Thread currentThread() getId() + : Tasks are paused );
break;
}
}
}
running = false;
}
if(killed) {
killed = false;
break;
}
}
}catch(InterruptedException e) {
return;
}
//System out println(Thread currentThread() getId() + : Killed );
}
}
//ThreadPool java
package polarman threadpool;
import java util Collection; import java util Iterator; import java util Vector;
/** *//**
*線程池
* @author ryang
*
*/
public class ThreadPool {
protected int maxPoolSize;
protected int initPoolSize;
protected Vector threads = new Vector();
protected boolean initialized = false;
protected boolean hasIdleThread = false;
public ThreadPool(int maxPoolSize int initPoolSize) { this maxPoolSize = maxPoolSize; this initPoolSize = initPoolSize;
}
public void init() {
initialized = true;
for(int i= ; iinitPoolSize; i++) {
PooledThread thread = new PooledThread(this);
thread start(); threads add(thread);
}
//System out println( 線程池初始化結(jié)束 線程數(shù)= + threads size() + 最大線程數(shù)= + maxPoolSize);
}
public void setMaxPoolSize(int maxPoolSize) { //System out println( 重設(shè)最大線程數(shù) 最大線程數(shù)= + maxPoolSize); this maxPoolSize = maxPoolSize;
if(maxPoolSize getPoolSize())
setPoolSize(maxPoolSize);
}
/** *//**
*重設(shè)當(dāng)前線程數(shù)
* 若需殺掉某線程 線程不會(huì)立刻殺掉 而會(huì)等到線程中的事務(wù)處理完成* 但此方法會(huì)立刻從線程池中移除該線程 不會(huì)等待事務(wù)處理結(jié)束
* @param size
*/
public void setPoolSize(int size) { if(!initialized) {
initPoolSize = size;
return;
}else if(size getPoolSize()) { for(int i=getPoolSize(); isize imaxPoolSize; i++) {
PooledThread thread = new PooledThread(this);
thread start(); threads add(thread);
}
}else if(size getPoolSize()) { while(getPoolSize() size) { PooledThread th = (PooledThread)threads remove( ); th kill();
}
}
//System out println( 重設(shè)線程數(shù) 線程數(shù)= + threads size());
}
public int getPoolSize() { return threads size();
}
protected void notifyForIdleThread() {
hasIdleThread = true;
}
protected boolean waitForIdleThread() {
hasIdleThread = false;
while(!hasIdleThread getPoolSize() = maxPoolSize) { try { Thread sleep( ); } catch (InterruptedException e) {
return false;
}
}
return true;
}
public synchronized PooledThread getIdleThread() { while(true) { for(Iterator itr=erator(); itr hasNext();) { PooledThread th = (PooledThread)itr next(); if(!th isRunning())
return th;
}
if(getPoolSize() maxPoolSize) {
PooledThread thread = new PooledThread(this);
thread start(); threads add(thread);
return thread;
}
//System out println( 線程池已滿(mǎn) 等待 );
if(waitForIdleThread() == false)
return null;
}
}
public void processTask(ThreadTask task) {
PooledThread th = getIdleThread();
if(th != null) { th putTask(task); th startTasks();
}
}
public void processTasksInSingleThread(ThreadTask[] tasks) {
PooledThread th = getIdleThread();
if(th != null) { th putTasks(tasks); th startTasks();
}
}
public void processTasksInSingleThread(Collection tasks) {
PooledThread th = getIdleThread();
if(th != null) { th putTasks(tasks); th startTasks();
}
}
}
下面是線程池的測(cè)試程序
//ThreadPoolTest java
import java io BufferedReader; import java io IOException; import java io InputStreamReader;
import polarman threadpool ThreadPool; import polarman threadpool ThreadTask;
public class ThreadPoolTest {
public static void main(String[] args) { System out println( quit 退出 ); System out println( task A 啟動(dòng)任務(wù)A 時(shí)長(zhǎng)為 秒 ); System out println( size 設(shè)置當(dāng)前線程池大小為 ); System out println( max 設(shè)置線程池最大線程數(shù)為 ); System out println();
final ThreadPool pool = new ThreadPool( ); pool init();
Thread cmdThread = new Thread() { public void run() {
BufferedReader reader = new BufferedReader(new InputStreamReader(System in));
while(true) { try { String line = reader readLine(); String words[] = line split( ); if(words[ ] equalsIgnoreCase( quit )) { System exit( ); }else if(words[ ] equalsIgnoreCase( size ) words length = ) { try { int size = Integer parseInt(words[ ]); pool setPoolSize(size); }catch(Exception e) {
}
}else if(words[ ] equalsIgnoreCase( max ) words length = ) { try { int max = Integer parseInt(words[ ]); pool setMaxPoolSize(max); }catch(Exception e) {
}
}else if(words[ ] equalsIgnoreCase( task ) words length = ) { try { int timelen = Integer parseInt(words[ ]); SimpleTask task = new SimpleTask(words[ ] timelen * ); pool processTask(task); }catch(Exception e) {
}
}
} catch (IOException e) { e printStackTrace();
}
}
}
};
cmdThread start();
/**//*
for(int i= ; i ; i++){
SimpleTask task = new SimpleTask( Task + i (i+ )* ); pool processTask(task);
}*/
}
}
class SimpleTask implements ThreadTask {
private String taskName;
private int timeLen;
public SimpleTask(String taskName int timeLen) { this taskName = taskName; this timeLen = timeLen;
}
public void run() { System out println(Thread currentThread() getId() +
: START TASK + taskName + );
try { Thread sleep(timeLen); } catch (InterruptedException e) {
}
System out println(Thread currentThread() getId() +
: END TASK + taskName + );
}
}
使用此線程池相當(dāng)簡(jiǎn)單 下面兩行代碼初始化線程池
ThreadPool pool = new ThreadPool( ); pool init();
要處理的任務(wù)實(shí)現(xiàn)ThreadTask 接口即可（如測(cè)試代碼里的SimpleTask） 這個(gè)接口只有一個(gè)方法run()
兩行代碼即可調(diào)用
lishixinzhi/Article/program/Java/hx/201311/27203
ava通過(guò)Executors提供四種線程池，分別為：
newCachedThreadPool創(chuàng)建一個(gè)可緩存線程池，如果線程池長(zhǎng)度超過(guò)處理需要，可靈活回收空閑線程，若無(wú)可回收，則新建線程。
newFixedThreadPool
創(chuàng)建一個(gè)定長(zhǎng)線程池，可控制線程最大并發(fā)數(shù)，超出的線程會(huì)在隊(duì)列中等待。
newScheduledThreadPool
創(chuàng)建一個(gè)定長(zhǎng)線程池，支持定時(shí)及周期性任務(wù)執(zhí)行。
newSingleThreadExecutor
創(chuàng)建一個(gè)單線程化的線程池，它只會(huì)用唯一的工作線程來(lái)執(zhí)行任務(wù)，保證所有任務(wù)按照指定順序(FIFO,
LIFO,
優(yōu)先級(jí))執(zhí)行。
線程池簡(jiǎn)介：
多線程技術(shù)主要解決處理器單元內(nèi)多個(gè)線程執(zhí)行的問(wèn)題，它可以顯著減少處理器單元的閑置時(shí)間，增加處理器單元的吞吐能力。
假設(shè)一個(gè)服務(wù)器完成一項(xiàng)任務(wù)所需時(shí)間為：T1 創(chuàng)建線程時(shí)間，T2 在線程中執(zhí)行任務(wù)的時(shí)間，T3 銷(xiāo)毀線程時(shí)間。
如果：T1 + T3 遠(yuǎn)大于 T2，則可以采用線程池，以提高服務(wù)器性能。
一個(gè)線程池包括以下四個(gè)基本組成部分：
1、線程池管理器（ThreadPool）：用于創(chuàng)建并管理線程池，包括 創(chuàng)建線程池，銷(xiāo)毀線程池，添加新任務(wù)；
2、工作線程（PoolWorker）：線程池中線程，在沒(méi)有任務(wù)時(shí)處于等待狀態(tài)，可以循環(huán)的執(zhí)行任務(wù)；
3、任務(wù)接口（Task）：每個(gè)任務(wù)必須實(shí)現(xiàn)的接口，以供工作線程調(diào)度任務(wù)的執(zhí)行，它主要規(guī)定了任務(wù)的入口，任務(wù)執(zhí)行完后的收尾工作，任務(wù)的執(zhí)行狀態(tài)等；
4、任務(wù)隊(duì)列（taskQueue）：用于存放沒(méi)有處理的任務(wù)。提供一種緩沖機(jī)制。
線程池技術(shù)正是關(guān)注如何縮短或調(diào)整T1,T3時(shí)間的技術(shù)，從而提高服務(wù)器程序性能的。它把T1，T3分別安排在服務(wù)器程序的啟動(dòng)和結(jié)束的時(shí)間段或者一些空閑的時(shí)間段，這樣在服務(wù)器程序處理客戶(hù)請(qǐng)求時(shí)，不會(huì)有T1，T3的開(kāi)銷(xiāo)了。
線程池不僅調(diào)整T1,T3產(chǎn)生的時(shí)間段，而且它還顯著減少了創(chuàng)建線程的數(shù)目，看一個(gè)例子：
假設(shè)一個(gè)服務(wù)器一天要處理50000個(gè)請(qǐng)求，并且每個(gè)請(qǐng)求需要一個(gè)單獨(dú)的線程完成。在線程池中，線程數(shù)一般是固定的，所以產(chǎn)生線程總數(shù)不會(huì)超過(guò)線程池中線程的數(shù)目，而如果服務(wù)器不利用線程池來(lái)處理這些請(qǐng)求則線程總數(shù)為50000。一般線程池大小是遠(yuǎn)小于50000。所以利用線程池的服務(wù)器程序不會(huì)為了創(chuàng)建50000而在處理請(qǐng)求時(shí)浪費(fèi)時(shí)間，從而提高效率。
代碼實(shí)現(xiàn)中并沒(méi)有實(shí)現(xiàn)任務(wù)接口，而是把Runnable對(duì)象加入到線程池管理器（ThreadPool），然后剩下的事情就由線程池管理器（ThreadPool）來(lái)完成了
package?mine.util.thread;??
import?java.util.LinkedList;??
import?java.util.List;??
/**?
*?線程池類(lèi)，線程管理器：創(chuàng)建線程，執(zhí)行任務(wù)，銷(xiāo)毀線程，獲取線程基本信息?
*/??
public?final?class?ThreadPool?{??
//?線程池中默認(rèn)線程的個(gè)數(shù)為5??
private?static?int?worker_num?=?5;??
//?工作線程??
private?WorkThread[]?workThrads;??
//?未處理的任務(wù)??
private?static?volatile?int?finished_task?=?0;??
//?任務(wù)隊(duì)列，作為一個(gè)緩沖,List線程不安全??
private?ListRunnable?taskQueue?=?new?LinkedListRunnable();??
private?static?ThreadPool?threadPool;??
//?創(chuàng)建具有默認(rèn)線程個(gè)數(shù)的線程池??
private?ThreadPool()?{??
this(5);??
}??
//?創(chuàng)建線程池,worker_num為線程池中工作線程的個(gè)數(shù)??
private?ThreadPool(int?worker_num)?{??
ThreadPool.worker_num?=?worker_num;??
workThrads?=?new?WorkThread[worker_num];??
for?(int?i?=?0;?i??worker_num;?i++)?{??
workThrads[i]?=?new?WorkThread();??
workThrads[i].start();//?開(kāi)啟線程池中的線程??
}??
}??
//?單態(tài)模式，獲得一個(gè)默認(rèn)線程個(gè)數(shù)的線程池??
public?static?ThreadPool?getThreadPool()?{??
return?getThreadPool(ThreadPool.worker_num);??
}??
//?單態(tài)模式，獲得一個(gè)指定線程個(gè)數(shù)的線程池,worker_num(0)為線程池中工作線程的個(gè)數(shù)??
//?worker_num=0創(chuàng)建默認(rèn)的工作線程個(gè)數(shù)??
public?static?ThreadPool?getThreadPool(int?worker_num1)?{??
if?(worker_num1?=?0)??
worker_num1?=?ThreadPool.worker_num;??
if?(threadPool?==?null)??
threadPool?=?new?ThreadPool(worker_num1);??
return?threadPool;??
}??
//?執(zhí)行任務(wù),其實(shí)只是把任務(wù)加入任務(wù)隊(duì)列，什么時(shí)候執(zhí)行有線程池管理器覺(jué)定??
public?void?execute(Runnable?task)?{??
synchronized?(taskQueue)?{??
taskQueue.add(task);??
taskQueue.notify();??
}??
}??
//?批量執(zhí)行任務(wù),其實(shí)只是把任務(wù)加入任務(wù)隊(duì)列，什么時(shí)候執(zhí)行有線程池管理器覺(jué)定??
public?void?execute(Runnable[]?task)?{??
synchronized?(taskQueue)?{??
for?(Runnable?t?:?task)??
taskQueue.add(t);??
taskQueue.notify();??
}??
}??
//?批量執(zhí)行任務(wù),其實(shí)只是把任務(wù)加入任務(wù)隊(duì)列，什么時(shí)候執(zhí)行有線程池管理器覺(jué)定??
public?void?execute(ListRunnable?task)?{??
synchronized?(taskQueue)?{??
for?(Runnable?t?:?task)??
taskQueue.add(t);??
taskQueue.notify();??
}??
}??
//?銷(xiāo)毀線程池,該方法保證在所有任務(wù)都完成的情況下才銷(xiāo)毀所有線程，否則等待任務(wù)完成才銷(xiāo)毀??
public?void?destroy()?{??
while?(!taskQueue.isEmpty())?{//?如果還有任務(wù)沒(méi)執(zhí)行完成，就先睡會(huì)吧??
try?{??
Thread.sleep(10);??
}?catch?(InterruptedException?e)?{??
e.printStackTrace();??
}??
}??
//?工作線程停止工作，且置為null??
for?(int?i?=?0;?i??worker_num;?i++)?{??
workThrads[i].stopWorker();??
workThrads[i]?=?null;??
}??
threadPool=null;??
taskQueue.clear();//?清空任務(wù)隊(duì)列??
}??
//?返回工作線程的個(gè)數(shù)??
public?int?getWorkThreadNumber()?{??
return?worker_num;??
}??
//?返回已完成任務(wù)的個(gè)數(shù),這里的已完成是只出了任務(wù)隊(duì)列的任務(wù)個(gè)數(shù)，可能該任務(wù)并沒(méi)有實(shí)際執(zhí)行完成??
public?int?getFinishedTasknumber()?{??
return?finished_task;??
}??
//?返回任務(wù)隊(duì)列的長(zhǎng)度，即還沒(méi)處理的任務(wù)個(gè)數(shù)??
public?int?getWaitTasknumber()?{??
return?taskQueue.size();??
}??
//?覆蓋toString方法，返回線程池信息：工作線程個(gè)數(shù)和已完成任務(wù)個(gè)數(shù)??
@Override??
public?String?toString()?{??
return?"WorkThread?number:"?+?worker_num?+?"??finished?task?number:"??
+?finished_task?+?"??wait?task?number:"?+?getWaitTasknumber();??
}??
/**?
*?內(nèi)部類(lèi)，工作線程?
*/??
private?class?WorkThread?extends?Thread?{??
//?該工作線程是否有效，用于結(jié)束該工作線程??
private?boolean?isRunning?=?true;??
/*?
*?關(guān)鍵所在啊，如果任務(wù)隊(duì)列不空，則取出任務(wù)執(zhí)行，若任務(wù)隊(duì)列空，則等待?
*/??
@Override??
public?void?run()?{??
Runnable?r?=?null;??
while?(isRunning)?{//?注意，若線程無(wú)效則自然結(jié)束run方法，該線程就沒(méi)用了??
synchronized?(taskQueue)?{??
while?(isRunning??taskQueue.isEmpty())?{//?隊(duì)列為空??
try?{??
taskQueue.wait(20);??
}?catch?(InterruptedException?e)?{??
e.printStackTrace();??
}??
}??
if?(!taskQueue.isEmpty())??
r?=?taskQueue.remove(0);//?取出任務(wù)??
}??
if?(r?!=?null)?{??
r.run();//?執(zhí)行任務(wù)??
}??
finished_task++;??
r?=?null;??
}??
}??
//?停止工作，讓該線程自然執(zhí)行完run方法，自然結(jié)束??
public?void?stopWorker()?{??
isRunning?=?false;??
}??
}??
}
最簡(jiǎn)單的可以利用java.util.concurrent.Executors
調(diào)用Executors.newCachedThreadPool()獲取緩沖式線程池
Executors.newFixedThreadPool(int nThreads)獲取固定大小的線程池
重慶分公司
重慶分公司
