Java手寫(xiě)線(xiàn)程池之向JDK線(xiàn)程池進(jìn)發(fā)

更新時(shí)間：2022年10月23日 10:06:46 作者：一無(wú)是處的研究僧

在前面的文章自己動(dòng)手寫(xiě)乞丐版線(xiàn)程池中，我們寫(xiě)了一個(gè)非常簡(jiǎn)單的線(xiàn)程池實(shí)現(xiàn)，這個(gè)只是一個(gè)非常簡(jiǎn)單的實(shí)現(xiàn)，在本篇文章當(dāng)中我們將要實(shí)現(xiàn)一個(gè)和JDK內(nèi)部實(shí)現(xiàn)的線(xiàn)程池非常相似的線(xiàn)程池，需要的可以了解一下

前言

JDK線(xiàn)程池一瞥

我們首先看一個(gè)JDK給我們提供的線(xiàn)程池ThreadPoolExecutor的構(gòu)造函數(shù)的參數(shù)：

public?ThreadPoolExecutor(int?corePoolSize,
??????????????????????????????int?maximumPoolSize,
??????????????????????????????long?keepAliveTime,
??????????????????????????????TimeUnit?unit,
??????????????????????????????BlockingQueue<Runnable>?workQueue,
??????????????????????????????ThreadFactory?threadFactory,
??????????????????????????????RejectedExecutionHandler?handler)?

參數(shù)解釋?zhuān)?/p>

1.corePoolSize：這個(gè)參數(shù)你可以理解為線(xiàn)程池當(dāng)中至少需要 corePoolSize 個(gè)線(xiàn)程，初始時(shí)線(xiàn)程池當(dāng)中線(xiàn)程的個(gè)數(shù)為0，當(dāng)線(xiàn)程池當(dāng)中線(xiàn)程的個(gè)數(shù)小于 corePoolSize 每次提交一個(gè)任務(wù)都會(huì)創(chuàng)建一個(gè)線(xiàn)程，并且先執(zhí)行這個(gè)提交的任務(wù)，然后再去任務(wù)隊(duì)列里面去獲取新的任務(wù)，然后再執(zhí)行。

2.maximumPoolSize：這個(gè)參數(shù)指的是線(xiàn)程池當(dāng)中能夠允許的最大的線(xiàn)程的數(shù)目，當(dāng)任務(wù)隊(duì)列滿(mǎn)了之后如果這個(gè)時(shí)候有新的任務(wù)想要加入隊(duì)列當(dāng)中，當(dāng)發(fā)現(xiàn)隊(duì)列滿(mǎn)了之后就創(chuàng)建新的線(xiàn)程去執(zhí)行任務(wù)，但是需要滿(mǎn)足最大的線(xiàn)程的個(gè)數(shù)不能夠超過(guò) maximumPoolSize 。

3.keepAliveTime 和 unit：這個(gè)主要是用于時(shí)間的表示，當(dāng)隊(duì)列當(dāng)中多長(zhǎng)時(shí)間沒(méi)有數(shù)據(jù)的時(shí)候線(xiàn)程自己退出，前面談到了線(xiàn)程池當(dāng)中任務(wù)過(guò)多的時(shí)候會(huì)超過(guò) corePoolSize ，當(dāng)線(xiàn)程池閑下來(lái)的時(shí)候這些多余的線(xiàn)程就可以退出了。

4.workQueue：這個(gè)就是用于保存任務(wù)的阻塞隊(duì)列。

5.threadFactory：這個(gè)參數(shù)倒不是很重要，線(xiàn)程工廠(chǎng)。

6.handler：這個(gè)表示拒絕策略，JDK給我們提供了四種策略：

AbortPolicy：拋出異常。
DiscardPolicy：放棄這個(gè)任務(wù)。
CallerRunPolicy：提交任務(wù)的線(xiàn)程執(zhí)行。
DiscardOldestPolicy:放棄等待時(shí)間最長(zhǎng)的任務(wù)。

如果上面的參數(shù)你不能夠理解，可以先閱讀這篇文章自己動(dòng)手寫(xiě)乞丐版線(xiàn)程池。基于上面談到的參數(shù)，線(xiàn)程池當(dāng)中提交任務(wù)的流程大致如下圖所示：

自己動(dòng)手實(shí)現(xiàn)線(xiàn)程池

根據(jù)前面的參數(shù)分析我們自己實(shí)現(xiàn)的線(xiàn)程池需要實(shí)現(xiàn)一下功能：

能夠提交Runnable的任務(wù)和Callable的任務(wù)。
線(xiàn)程池能夠自己實(shí)現(xiàn)動(dòng)態(tài)的擴(kuò)容和所容，動(dòng)態(tài)調(diào)整線(xiàn)程池當(dāng)中線(xiàn)程的數(shù)目，當(dāng)任務(wù)多的時(shí)候能夠增加線(xiàn)程的數(shù)目，當(dāng)任務(wù)少的時(shí)候多出來(lái)的線(xiàn)程能夠自動(dòng)退出。
有自己的拒絕策略，當(dāng)任務(wù)隊(duì)列滿(mǎn)了，線(xiàn)程數(shù)也達(dá)到最大的時(shí)候，需要拒絕提交的任務(wù)。

線(xiàn)程池參數(shù)介紹

??private?AtomicInteger?ct?=?new?AtomicInteger(0);?//?當(dāng)前在執(zhí)行任務(wù)的線(xiàn)程個(gè)數(shù)
??private?int?corePoolSize;
??private?int?maximumPoolSize;
??private?long?keepAliveTime;
??private?TimeUnit?unit;
??private?BlockingQueue<Runnable>?taskQueue;
??private?RejectPolicy?policy;

??private?ArrayList<Worker>?workers?=?new?ArrayList<>();

??private?volatile?boolean?isStopped;
??private?boolean?useTimed;

參數(shù)解釋如下：

ct：表示當(dāng)前線(xiàn)程池當(dāng)中線(xiàn)程的個(gè)數(shù)。
corePoolSize：線(xiàn)程池當(dāng)中核心線(xiàn)程的個(gè)數(shù)，意義和上面談到的JDK的線(xiàn)程池意義一致。
maximumPoolSize：線(xiàn)程池當(dāng)中最大的線(xiàn)程個(gè)數(shù)，意義和上面談到的JDK的線(xiàn)程池意義一致。
keepAliveTime 和 unit：和JDK線(xiàn)程池的參數(shù)意義一致。
taskQueue：任務(wù)隊(duì)列，用不保存提交的任務(wù)。
policy：拒絕策略，主要有一下四種策略：

public?enum?RejectPolicy?{

??ABORT,
??CALLER_RUN,
??DISCARD_OLDEST,
??DISCARD
}

workers：用于保存工作線(xiàn)程。

isStopped：線(xiàn)程池是否被關(guān)閉了。

useTimed：主要是用于表示是否使用上面的 keepAliveTime 和 unit，如果使用就是在一定的時(shí)間內(nèi)，如果沒(méi)有從任務(wù)隊(duì)列當(dāng)中獲取到任務(wù)，線(xiàn)程就從線(xiàn)程池退出，但是需要保證線(xiàn)程池當(dāng)中最小的線(xiàn)程個(gè)數(shù)不小于 corePoolSize 。

實(shí)現(xiàn)Runnable

??//?下面這個(gè)方法是向線(xiàn)程池提交任務(wù)
??public?void?execute(Runnable?runnable)?throws?InterruptedException?{
????checkPoolState();

????if?(addWorker(runnable,?false)??//?如果能夠加入新的線(xiàn)程執(zhí)行任務(wù)?加入成功就直接返回
????????????||?!taskQueue.offer(runnable)?//?如果?taskQueue.offer(runnable)?返回?false?說(shuō)明提交任務(wù)失敗?任務(wù)隊(duì)列已經(jīng)滿(mǎn)了
????????????||?addWorker(runnable,?true))?//?使用能夠使用的最大的線(xiàn)程數(shù)?(maximumPoolSize)?看是否能夠產(chǎn)生新的線(xiàn)程
??????return;

????//?如果任務(wù)隊(duì)列滿(mǎn)了而且不能夠加入新的線(xiàn)程?則拒絕這個(gè)任務(wù)
????if?(!taskQueue.offer(runnable))
??????reject(runnable);
??}

在上面的代碼當(dāng)中：

checkPoolState函數(shù)是檢查線(xiàn)程池的狀態(tài)，當(dāng)線(xiàn)程池被停下來(lái)之后就不能夠在提交任務(wù)：

??private?void?checkPoolState()?{
????if?(isStopped)?{
??????//?如果線(xiàn)程池已經(jīng)停下來(lái)了，就不在向任務(wù)隊(duì)列當(dāng)中提交任務(wù)了
??????throw?new?RuntimeException("thread?pool?has?been?stopped,?so?quit?submitting?task");
????}
??}

addWorker函數(shù)是往線(xiàn)程池當(dāng)中提交任務(wù)并且產(chǎn)生一個(gè)線(xiàn)程，并且這個(gè)線(xiàn)程執(zhí)行的第一個(gè)任務(wù)就是傳遞的參數(shù)。max表示線(xiàn)程的最大數(shù)目，max == true 的時(shí)候表示使用 maximumPoolSize 否則使用 corePoolSize，當(dāng)返回值等于 true 的時(shí)候表示執(zhí)行成功，否則表示執(zhí)行失敗。

??/**
???*
???*?@param?runnable?需要被執(zhí)行的任務(wù)
???*?@param?max?是否使用?maximumPoolSize
???*?@return?boolean
???*/
??public?synchronized?boolean?addWorker(Runnable?runnable,?boolean?max)?{

????if?(ct.get()?>=?corePoolSize?&&?!max)
??????return?false;
????if?(ct.get()?>=?maximumPoolSize?&&?max)
??????return?false;
????Worker?worker?=?new?Worker(runnable);
????workers.add(worker);
????Thread?thread?=?new?Thread(worker,?"ThreadPool-"?+?"Thread-"?+?ct.addAndGet(1));
????thread.start();
????return?true;
??}

實(shí)現(xiàn)Callable

這個(gè)函數(shù)其實(shí)比較簡(jiǎn)單，只需要將傳入的Callable對(duì)象封裝成一個(gè)FutureTask對(duì)象即可，因?yàn)镕utureTask實(shí)現(xiàn)了Callable和Runnable兩個(gè)接口，然后將這個(gè)結(jié)果返回即可，得到這個(gè)對(duì)象，再調(diào)用對(duì)象的 get 方法就能夠得到結(jié)果。

??public?<V>?RunnableFuture<V>?submit(Callable<V>?task)?throws?InterruptedException?{
????checkPoolState();
????FutureTask<V>?futureTask?=?new?FutureTask<>(task);
????execute(futureTask);
????return?futureTask;
??}

拒絕策略的實(shí)現(xiàn)

根據(jù)前面提到的各種策略的具體實(shí)現(xiàn)方式，具體的代碼實(shí)現(xiàn)如下所示：

??private?void?reject(Runnable?runnable)?throws?InterruptedException?{
????switch?(policy)?{
??????case?ABORT:
????????throw?new?RuntimeException("task?queue?is?full");
??????case?CALLER_RUN:
????????runnable.run();
??????case?DISCARD:?//?直接放棄這個(gè)任務(wù)
????????return;
??????case?DISCARD_OLDEST:
????????//?放棄等待時(shí)間最長(zhǎng)的任務(wù)?也就是隊(duì)列當(dāng)中的第一個(gè)任務(wù)
????????taskQueue.poll();
????????execute(runnable);?//?重新執(zhí)行這個(gè)任務(wù)
????}
??}

線(xiàn)程池關(guān)閉實(shí)現(xiàn)

一共兩種方式實(shí)現(xiàn)線(xiàn)程池關(guān)閉：

直接關(guān)閉線(xiàn)程池，不管任務(wù)隊(duì)列當(dāng)中的任務(wù)是否被全部執(zhí)行完成。
安全關(guān)閉線(xiàn)程池，先等待任務(wù)隊(duì)列當(dāng)中所有的任務(wù)被執(zhí)行完成，再關(guān)閉線(xiàn)程池，但是在這個(gè)過(guò)程當(dāng)中不允許繼續(xù)提交任務(wù)了，這一點(diǎn)已經(jīng)在函數(shù) checkPoolState 當(dāng)中實(shí)現(xiàn)了。

??//?強(qiáng)制關(guān)閉線(xiàn)程池
??public?synchronized?void?stop()?{
????isStopped?=?true;
????for?(Worker?worker?:?workers)?{
??????worker.stopWorker();
????}
??}

??public?synchronized?void?shutDown()?{
????//?先表示關(guān)閉線(xiàn)程池?線(xiàn)程就不能再向線(xiàn)程池提交任務(wù)
????isStopped?=?true;
????//?先等待所有的任務(wù)執(zhí)行完成再關(guān)閉線(xiàn)程池
????waitForAllTasks();
????stop();
??}

??private?void?waitForAllTasks()?{
????//?當(dāng)線(xiàn)程池當(dāng)中還有任務(wù)的時(shí)候?就不退出循環(huán)
????while?(taskQueue.size()?>?0)?{
??????Thread.yield();
??????try?{
????????Thread.sleep(1000);
??????}?catch?(InterruptedException?e)?{
????????e.printStackTrace();
??????}
????}
??}

工作線(xiàn)程的工作實(shí)現(xiàn)

????@Override
????public?void?run()?{
??????//?先執(zhí)行傳遞過(guò)來(lái)的第一個(gè)任務(wù)?這里是一個(gè)小的優(yōu)化?讓線(xiàn)程直接執(zhí)行第一個(gè)任務(wù)?不需要
??????//?放入任務(wù)隊(duì)列再取出來(lái)執(zhí)行了
??????firstTask.run();

??????thisThread?=?Thread.currentThread();
??????while?(!isStopped)?{
????????try?{
??????????//?是否使用時(shí)間就在這里顯示出來(lái)了
??????????Runnable?task?=?useTimed???taskQueue.poll(keepAliveTime,?unit)?:?taskQueue.take();
??????????if?(task?==?null)?{
????????????int?i;
????????????boolean?exit?=?true;
????????????//?如果當(dāng)前線(xiàn)程數(shù)大于核心線(xiàn)程數(shù)?則使用?CAS?去退出?用于保證在線(xiàn)程安全下的退出
????????????//?且保證線(xiàn)程的個(gè)數(shù)不小于?corePoolSize?下面這段代碼需要仔細(xì)分析一下
????????????if?(ct.get()?>?corePoolSize)?{
??????????????do{
????????????????i?=?ct.get();
????????????????if?(i?<=?corePoolSize)?{
??????????????????exit?=?false;
??????????????????break;
????????????????}
??????????????}while?(!ct.compareAndSet(i,?i?-?1));
??????????????if?(exit)?{
????????????????return;
??????????????}
????????????}
??????????}else?{
????????????task.run();
??????????}
????????}?catch?(InterruptedException?e)?{
??????????//?do?nothing
????????}
??????}
????}

我們現(xiàn)在來(lái)仔細(xì)分析一下，線(xiàn)程退出線(xiàn)程池的時(shí)候是如何保證線(xiàn)程池當(dāng)中總的線(xiàn)程數(shù)是不小于 corePoolSize 的！首先整體的框架是使用 CAS 進(jìn)行實(shí)現(xiàn)，具體代碼為 do ... while 操作，然后在 while 操作里面使用 CAS 進(jìn)行測(cè)試替換，如果沒(méi)有成功再次獲取，當(dāng)線(xiàn)程池當(dāng)中核心線(xiàn)程的數(shù)目小于等于 corePoolSize 的時(shí)候也需要退出循環(huán)，因?yàn)榫€(xiàn)程池當(dāng)中線(xiàn)程的個(gè)數(shù)不能小于 corePoolSize 。因此使用 break 跳出循環(huán)的線(xiàn)程是不會(huì)退出線(xiàn)程池的。

線(xiàn)程池實(shí)現(xiàn)的BUG

在我們自己實(shí)現(xiàn)的線(xiàn)程池當(dāng)中當(dāng)線(xiàn)程退出的時(shí)候，workers 當(dāng)中還保存這指向這個(gè)線(xiàn)程的對(duì)象，但是當(dāng)線(xiàn)程退出的時(shí)候我們還沒(méi)有在 workers 當(dāng)中刪除這個(gè)對(duì)象，因此這個(gè)線(xiàn)程對(duì)象不會(huì)被垃圾回收器收集掉，但是我們這個(gè)只是一個(gè)線(xiàn)程池實(shí)現(xiàn)的例子而已，并不用于生產(chǎn)環(huán)境，只是為了幫助大家理解線(xiàn)程池的原理。

完整代碼

package?cscore.concurrent.java.threadpoolv2;


import?java.util.ArrayList;
import?java.util.concurrent.*;
import?java.util.concurrent.atomic.AtomicInteger;

public?class?ThreadPool?{

??private?AtomicInteger?ct?=?new?AtomicInteger(0);?//?當(dāng)前在執(zhí)行任務(wù)的線(xiàn)程個(gè)數(shù)
??private?int?corePoolSize;
??private?int?maximumPoolSize;
??private?long?keepAliveTime;
??private?TimeUnit?unit;
??private?BlockingQueue<Runnable>?taskQueue;
??private?RejectPolicy?policy;

??private?ArrayList<Worker>?workers?=?new?ArrayList<>();

??private?volatile?boolean?isStopped;
??private?boolean?useTimed;

??public?int?getCt()?{
????return?ct.get();
??}

??public?ThreadPool(int?corePoolSize,?int?maximumPoolSize,?TimeUnit?unit,?long?keepAliveTime,?RejectPolicy?policy
??????????,?int?maxTasks)?{
????//?please?add?-ea?to?vm?options?to?make?assert?keyword?enable
????assert?corePoolSize?>?0;
????assert?maximumPoolSize?>?0;
????assert?keepAliveTime?>=?0;
????assert?maxTasks?>?0;

????this.corePoolSize?=?corePoolSize;
????this.maximumPoolSize?=?maximumPoolSize;
????this.unit?=?unit;
????this.policy?=?policy;
????this.keepAliveTime?=?keepAliveTime;
????taskQueue?=?new?ArrayBlockingQueue<Runnable>(maxTasks);
????useTimed?=?keepAliveTime?!=?0;
??}

??/**
???*
???*?@param?runnable?需要被執(zhí)行的任務(wù)
???*?@param?max?是否使用?maximumPoolSize
???*?@return?boolean
???*/
??public?synchronized?boolean?addWorker(Runnable?runnable,?boolean?max)?{

????if?(ct.get()?>=?corePoolSize?&&?!max)
??????return?false;
????if?(ct.get()?>=?maximumPoolSize?&&?max)
??????return?false;
????Worker?worker?=?new?Worker(runnable);
????workers.add(worker);
????Thread?thread?=?new?Thread(worker,?"ThreadPool-"?+?"Thread-"?+?ct.addAndGet(1));
????thread.start();
????return?true;
??}

??//?下面這個(gè)方法是向線(xiàn)程池提交任務(wù)
??public?void?execute(Runnable?runnable)?throws?InterruptedException?{
????checkPoolState();

????if?(addWorker(runnable,?false)??//?如果能夠加入新的線(xiàn)程執(zhí)行任務(wù)?加入成功就直接返回
????????????||?!taskQueue.offer(runnable)?//?如果?taskQueue.offer(runnable)?返回?false?說(shuō)明提交任務(wù)失敗?任務(wù)隊(duì)列已經(jīng)滿(mǎn)了
????????????||?addWorker(runnable,?true))?//?使用能夠使用的最大的線(xiàn)程數(shù)?(maximumPoolSize)?看是否能夠產(chǎn)生新的線(xiàn)程
??????return;

????//?如果任務(wù)隊(duì)列滿(mǎn)了而且不能夠加入新的線(xiàn)程?則拒絕這個(gè)任務(wù)
????if?(!taskQueue.offer(runnable))
??????reject(runnable);
??}

??private?void?reject(Runnable?runnable)?throws?InterruptedException?{
????switch?(policy)?{
??????case?ABORT:
????????throw?new?RuntimeException("task?queue?is?full");
??????case?CALLER_RUN:
????????runnable.run();
??????case?DISCARD:
????????return;
??????case?DISCARD_OLDEST:
????????//?放棄等待時(shí)間最長(zhǎng)的任務(wù)
????????taskQueue.poll();
????????execute(runnable);
????}
??}

??private?void?checkPoolState()?{
????if?(isStopped)?{
??????//?如果線(xiàn)程池已經(jīng)停下來(lái)了，就不在向任務(wù)隊(duì)列當(dāng)中提交任務(wù)了
??????throw?new?RuntimeException("thread?pool?has?been?stopped,?so?quit?submitting?task");
????}
??}

??public?<V>?RunnableFuture<V>?submit(Callable<V>?task)?throws?InterruptedException?{
????checkPoolState();
????FutureTask<V>?futureTask?=?new?FutureTask<>(task);
????execute(futureTask);
????return?futureTask;
??}

??//?強(qiáng)制關(guān)閉線(xiàn)程池
??public?synchronized?void?stop()?{
????isStopped?=?true;
????for?(Worker?worker?:?workers)?{
??????worker.stopWorker();
????}
??}

??public?synchronized?void?shutDown()?{
????//?先表示關(guān)閉線(xiàn)程池?線(xiàn)程就不能再向線(xiàn)程池提交任務(wù)
????isStopped?=?true;
????//?先等待所有的任務(wù)執(zhí)行完成再關(guān)閉線(xiàn)程池
????waitForAllTasks();
????stop();
??}

??private?void?waitForAllTasks()?{
????//?當(dāng)線(xiàn)程池當(dāng)中還有任務(wù)的時(shí)候?就不退出循環(huán)
????while?(taskQueue.size()?>?0)?{
??????Thread.yield();
??????try?{
????????Thread.sleep(1000);
??????}?catch?(InterruptedException?e)?{
????????e.printStackTrace();
??????}
????}
??}

??class?Worker?implements?Runnable?{

????private?Thread?thisThread;

????private?final?Runnable?firstTask;
????private?volatile?boolean?isStopped;

????public?Worker(Runnable?firstTask)?{
??????this.firstTask?=?firstTask;
????}

????@Override
????public?void?run()?{
??????//?先執(zhí)行傳遞過(guò)來(lái)的第一個(gè)任務(wù)?這里是一個(gè)小的優(yōu)化?讓線(xiàn)程直接執(zhí)行第一個(gè)任務(wù)?不需要
??????//?放入任務(wù)隊(duì)列再取出來(lái)執(zhí)行了
??????firstTask.run();

??????thisThread?=?Thread.currentThread();
??????while?(!isStopped)?{
????????try?{
??????????Runnable?task?=?useTimed???taskQueue.poll(keepAliveTime,?unit)?:?taskQueue.take();
??????????if?(task?==?null)?{
????????????int?i;
????????????boolean?exit?=?true;
????????????if?(ct.get()?>?corePoolSize)?{
??????????????do{
????????????????i?=?ct.get();
????????????????if?(i?<=?corePoolSize)?{
??????????????????exit?=?false;
??????????????????break;
????????????????}
??????????????}while?(!ct.compareAndSet(i,?i?-?1));
??????????????if?(exit)?{
????????????????return;
??????????????}
????????????}
??????????}else?{
????????????task.run();
??????????}
????????}?catch?(InterruptedException?e)?{
??????????//?do?nothing
????????}
??????}
????}

????public?synchronized?void?stopWorker()?{
??????if?(isStopped)?{
????????throw?new?RuntimeException("thread?has?been?interrupted");
??????}
??????isStopped?=?true;
??????thisThread.interrupt();
????}

??}

}

線(xiàn)程池測(cè)試

package?cscore.concurrent.java.threadpoolv2;

import?java.util.concurrent.ExecutionException;
import?java.util.concurrent.RunnableFuture;
import?java.util.concurrent.TimeUnit;

public?class?Test?{

??public?static?void?main(String[]?args)?throws?InterruptedException,?ExecutionException?{
????var?pool?=?new?ThreadPool(2,?5,?TimeUnit.SECONDS,?10,?RejectPolicy.ABORT,?100000);

????for?(int?i?=?0;?i?<?10;?i++)?{
??????RunnableFuture<Integer>?submit?=?pool.submit(()?->?{
????????System.out.println(Thread.currentThread().getName()?+?"?output?a");
????????try?{
??????????Thread.sleep(10);
????????}?catch?(InterruptedException?e)?{
??????????e.printStackTrace();
????????}
????????return?0;
??????});
??????System.out.println(submit.get());
????}
????int?n?=?15;
????while?(n--?>?0)?{
??????System.out.println("Number?Threads?=?"?+?pool.getCt());
??????Thread.sleep(1000);
????}
????pool.shutDown();
??}
}

上面測(cè)試代碼的輸出結(jié)果如下所示：

ThreadPool-Thread-2 output a
ThreadPool-Thread-1 output a
ThreadPool-Thread-3 output a
ThreadPool-Thread-4 output a
Number Threads = 5
ThreadPool-Thread-5 output a
ThreadPool-Thread-2 output a
ThreadPool-Thread-1 output a
ThreadPool-Thread-3 output a
ThreadPool-Thread-4 output a
ThreadPool-Thread-5 output a
ThreadPool-Thread-2 output a
ThreadPool-Thread-1 output a
ThreadPool-Thread-4 output a
ThreadPool-Thread-3 output a
ThreadPool-Thread-5 output a
ThreadPool-Thread-2 output a
ThreadPool-Thread-1 output a
ThreadPool-Thread-4 output a
Number Threads = 5
Number Threads = 5
Number Threads = 5
Number Threads = 5
Number Threads = 5
Number Threads = 5
Number Threads = 5
Number Threads = 5
Number Threads = 5
Number Threads = 3
Number Threads = 2
Number Threads = 2
Number Threads = 2
Number Threads = 2

從上面的代碼可以看出我們實(shí)現(xiàn)了正確的任務(wù)實(shí)現(xiàn)結(jié)果，同時(shí)線(xiàn)程池當(dāng)中的核心線(xiàn)程數(shù)從 2 變到了 5 ，當(dāng)線(xiàn)程池當(dāng)中任務(wù)隊(duì)列全部別執(zhí)行完成之后，線(xiàn)程的數(shù)目重新降下來(lái)了，這確實(shí)是我們想要達(dá)到的結(jié)果。