轉(zhuǎn)載自??Java線程池，從使用到原理

線程池的技術(shù)背景

在面向?qū)ο缶幊讨?#xff0c;創(chuàng)建和銷毀對(duì)象是很費(fèi)時(shí)間的，因?yàn)閯?chuàng)建一個(gè)對(duì)象要獲取內(nèi)存資源或者其它更多資源。在Java中更是如此，虛擬機(jī)將試圖跟蹤每一個(gè)對(duì)象，以便能夠在對(duì)象銷毀后進(jìn)行垃圾回收。

所以提高服務(wù)程序效率的一個(gè)手段就是盡可能減少創(chuàng)建和銷毀對(duì)象的次數(shù)，特別是一些很耗資源的對(duì)象創(chuàng)建和銷毀。如何利用已有對(duì)象來(lái)服務(wù)就是一個(gè)需要解決的關(guān)鍵問(wèn)題，其實(shí)這就是一些”池化資源”技術(shù)產(chǎn)生的原因。

例如Android中常見(jiàn)到的很多通用組件一般都離不開(kāi)”池”的概念，如各種圖片加載庫(kù)，網(wǎng)絡(luò)請(qǐng)求庫(kù)，即使Android的消息傳遞機(jī)制中的Meaasge當(dāng)使用Meaasge.obtain()就是使用的Meaasge池中的對(duì)象，因此這個(gè)概念很重要。本文將介紹的線程池技術(shù)同樣符合這一思想。

線程池的優(yōu)點(diǎn):

1.重用線程池中的線程,減少因?qū)ο髣?chuàng)建,銷毀所帶來(lái)的性能開(kāi)銷;

2.能有效的控制線程的最大并發(fā)數(shù),提高系統(tǒng)資源利用率,同時(shí)避免過(guò)多的資源競(jìng)爭(zhēng),避免堵塞;

3.能夠多線程進(jìn)行簡(jiǎn)單的管理,使線程的使用簡(jiǎn)單、高效。

線程池框架Executor

java中的線程池是通過(guò)Executor框架實(shí)現(xiàn)的，Executor 框架包括類：Executor，Executors，ExecutorService，ThreadPoolExecutor ，Callable和Future、FutureTask的使用等。

Executor: 所有線程池的接口,只有一個(gè)方法。

public interface Executor {void execute(Runnable command); }

ExecutorService: 增加Executor的行為，是Executor實(shí)現(xiàn)類的最直接接口。

Executors： 提供了一系列工廠方法用于創(chuàng)先線程池，返回的線程池都實(shí)現(xiàn)了ExecutorService 接口。

ThreadPoolExecutor：線程池的具體實(shí)現(xiàn)類,一般用的各種線程池都是基于這個(gè)類實(shí)現(xiàn)的。 構(gòu)造方法如下:

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }

corePoolSize：線程池的核心線程數(shù),線程池中運(yùn)行的線程數(shù)也永遠(yuǎn)不會(huì)超過(guò) corePoolSize個(gè),默認(rèn)情況下可以一直存活。可以通過(guò)設(shè)置allowCoreThreadTimeOut為True,此時(shí)核心線程數(shù)就是0,此時(shí)keepAliveTime控制所有線程的超時(shí)時(shí)間。

maximumPoolSize：線程池允許的最大線程數(shù);

keepAliveTime： 指的是空閑線程結(jié)束的超時(shí)時(shí)間;

unit ：是一個(gè)枚舉，表示 keepAliveTime 的單位;

workQueue：表示存放任務(wù)的BlockingQueue<Runnable>隊(duì)列。

BlockingQueue:阻塞隊(duì)列（BlockingQueue）是java.util.concurrent下的主要用來(lái)控制線程同步的工具。如果BlockQueue是空的,從BlockingQueue取東西的操作將會(huì)被阻斷進(jìn)入等待狀態(tài),直到BlockingQueue進(jìn)了東西才會(huì)被喚醒。同樣,如果BlockingQueue是滿的,任何試圖往里存東西的操作也會(huì)被阻斷進(jìn)入等待狀態(tài),直到BlockingQueue里有空間才會(huì)被喚醒繼續(xù)操作。 阻塞隊(duì)列常用于生產(chǎn)者和消費(fèi)者的場(chǎng)景，生產(chǎn)者是往隊(duì)列里添加元素的線程，消費(fèi)者是從隊(duì)列里拿元素的線程。阻塞隊(duì)列就是生產(chǎn)者存放元素的容器，而消費(fèi)者也只從容器里拿元素。具體的實(shí)現(xiàn)類有LinkedBlockingQueue，ArrayBlockingQueued等。一般其內(nèi)部的都是通過(guò)Lock和Condition來(lái)實(shí)現(xiàn)阻塞和喚醒。

線程池的工作過(guò)程如下：

線程池剛創(chuàng)建時(shí)，里面沒(méi)有一個(gè)線程。任務(wù)隊(duì)列是作為參數(shù)傳進(jìn)來(lái)的。不過(guò)，就算隊(duì)列里面有任務(wù)，線程池也不會(huì)馬上執(zhí)行它們。

當(dāng)調(diào)用 execute() 方法添加一個(gè)任務(wù)時(shí)，線程池會(huì)做如下判斷：

如果正在運(yùn)行的線程數(shù)量小于 corePoolSize，那么馬上創(chuàng)建線程運(yùn)行這個(gè)任務(wù)；

如果正在運(yùn)行的線程數(shù)量大于或等于 corePoolSize，那么將這個(gè)任務(wù)放入隊(duì)列；

如果這時(shí)候隊(duì)列滿了，而且正在運(yùn)行的線程數(shù)量小于 maximumPoolSize，那么還是要?jiǎng)?chuàng)建非核心線程立刻運(yùn)行這個(gè)任務(wù)；

如果隊(duì)列滿了，而且正在運(yùn)行的線程數(shù)量大于或等于 maximumPoolSize，那么線程池會(huì)拋出異常RejectExecutionException。

當(dāng)一個(gè)線程完成任務(wù)時(shí)，它會(huì)從隊(duì)列中取下一個(gè)任務(wù)來(lái)執(zhí)行。

當(dāng)一個(gè)線程無(wú)事可做，超過(guò)一定的時(shí)間（keepAliveTime）時(shí)，線程池會(huì)判斷，如果當(dāng)前運(yùn)行的線程數(shù)大于 corePoolSize，那么這個(gè)線程就被停掉。所以線程池的所有任務(wù)完成后，它最終會(huì)收縮到 corePoolSize 的大小。

線程池的創(chuàng)建和使用

生成線程池采用了工具類Executors的靜態(tài)方法，以下是幾種常見(jiàn)的線程池。

SingleThreadExecutor：單個(gè)后臺(tái)線程 (其緩沖隊(duì)列是無(wú)界的)

public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>())); }

創(chuàng)建一個(gè)單線程的線程池。這個(gè)線程池只有一個(gè)核心線程在工作，也就是相當(dāng)于單線程串行執(zhí)行所有任務(wù)。如果這個(gè)唯一的線程因?yàn)楫惓＝Y(jié)束，那么會(huì)有一個(gè)新的線程來(lái)替代它。此線程池保證所有任務(wù)的執(zhí)行順序按照任務(wù)的提交順序執(zhí)行。

FixedThreadPool：只有核心線程的線程池,大小固定 (其緩沖隊(duì)列是無(wú)界的) 。

public static ExecutorService newFixedThreadPool(int nThreads){return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()); }

創(chuàng)建固定大小的線程池。每次提交一個(gè)任務(wù)就創(chuàng)建一個(gè)線程，直到線程達(dá)到線程池的最大大小。線程池的大小一旦達(dá)到最大值就會(huì)保持不變，如果某個(gè)線程因?yàn)閳?zhí)行異常而結(jié)束，那么線程池會(huì)補(bǔ)充一個(gè)新線程。

CachedThreadPool：無(wú)界線程池，可以進(jìn)行自動(dòng)線程回收。

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>()); }

如果線程池的大小超過(guò)了處理任務(wù)所需要的線程，那么就會(huì)回收部分空閑（60秒不執(zhí)行任務(wù)）的線程，當(dāng)任務(wù)數(shù)增加時(shí)，此線程池又可以智能的添加新線程來(lái)處理任務(wù)。此線程池不會(huì)對(duì)線程池大小做限制，線程池大小完全依賴于操作系統(tǒng)（或者說(shuō)JVM）能夠創(chuàng)建的最大線程大小。SynchronousQueue是一個(gè)是緩沖區(qū)為1的阻塞隊(duì)列。

ScheduledThreadPool：核心線程池固定，大小無(wú)限的線程池。此線程池支持定時(shí)以及周期性執(zhí)行任務(wù)的需求。

public static ScheduledExecutorServicenewScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize); }

創(chuàng)建一個(gè)周期性執(zhí)行任務(wù)的線程池。如果閑置,非核心線程池會(huì)在DEFAULT_KEEPALIVEMILLIS時(shí)間內(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)的實(shí)現(xiàn)，submit(Runnable runnable)同理。(AbstractExecutorService.java)

public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask; }

可以看出submit開(kāi)啟的是有返回結(jié)果的任務(wù)，會(huì)返回一個(gè)FutureTask對(duì)象，這樣就能通過(guò)get()方法得到結(jié)果。submit最終調(diào)用的也是execute(Runnable runable)，submit只是將Callable對(duì)象或Runnable封裝成一個(gè)FutureTask對(duì)象，因?yàn)镕utureTask是個(gè)Runnable，所以可以在execute中執(zhí)行。關(guān)于Callable對(duì)象和Runnable怎么封裝成FutureTask對(duì)象，見(jiàn)Callable和Future、FutureTask的使用。

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

如果只講線程池的使用，那這篇博客沒(méi)有什么大的價(jià)值，充其量也就是熟悉Executor相關(guān)API的過(guò)程。線程池的實(shí)現(xiàn)過(guò)程沒(méi)有用到Synchronized關(guān)鍵字，用的都是Volatile，Lock和同步(阻塞)隊(duì)列，Atomic相關(guān)類，FutureTask等等，因?yàn)楹笳叩男阅芨鼉?yōu)。理解的過(guò)程可以很好的學(xué)習(xí)源碼中并發(fā)控制的思想。

在開(kāi)篇提到過(guò)線程池的優(yōu)點(diǎn)是可總結(jié)為以下三點(diǎn)：

線程復(fù)用

控制最大并發(fā)數(shù)

管理線程

1.線程復(fù)用過(guò)程

理解線程復(fù)用原理首先應(yīng)了解線程生命周期。

在線程的生命周期中，它要經(jīng)過(guò)新建(New)、就緒（Runnable）、運(yùn)行（Running）、阻塞(Blocked)和死亡(Dead)5種狀態(tài)。

Thread通過(guò)new來(lái)新建一個(gè)線程，這個(gè)過(guò)程是是初始化一些線程信息，如線程名，id，線程所屬group等，可以認(rèn)為只是個(gè)普通的對(duì)象。調(diào)用Thread的start()后Java虛擬機(jī)會(huì)為其創(chuàng)建方法調(diào)用棧和程序計(jì)數(shù)器，同時(shí)將hasBeenStarted為true，之后調(diào)用start方法就會(huì)有異常。

處于這個(gè)狀態(tài)中的線程并沒(méi)有開(kāi)始運(yùn)行，只是表示該線程可以運(yùn)行了。至于該線程何時(shí)開(kāi)始運(yùn)行，取決于JVM里線程調(diào)度器的調(diào)度。當(dāng)線程獲取CPU后，run()方法會(huì)被調(diào)用。不要自己去調(diào)用Thread的run()方法。之后根據(jù)CPU的調(diào)度在就緒——運(yùn)行——阻塞間切換，直到run()方法結(jié)束或其他方式停止線程，進(jìn)入dead狀態(tài)。

所以實(shí)現(xiàn)線程復(fù)用的原理應(yīng)該就是要保持線程處于存活狀態(tài)（就緒，運(yùn)行或阻塞）。接下來(lái)來(lái)看下ThreadPoolExecutor是怎么實(shí)現(xiàn)線程復(fù)用的。

在ThreadPoolExecutor主要Worker類來(lái)控制線程的復(fù)用。看下Worker類簡(jiǎn)化后的代碼，這樣方便理解：

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是一個(gè)Runnable，同時(shí)擁有一個(gè)thread，這個(gè)thread就是要開(kāi)啟的線程，在新建Worker對(duì)象時(shí)同時(shí)新建一個(gè)Thread對(duì)象，同時(shí)將Worker自己作為參數(shù)傳入TThread，這樣當(dāng)Thread的start()方法調(diào)用時(shí)，運(yùn)行的實(shí)際上是Worker的run()方法，接著到runWorker()中,有個(gè)while循環(huán)，一直從getTask()里得到Runnable對(duì)象，順序執(zhí)行。getTask()又是怎么得到Runnable對(duì)象的呢？

依舊是簡(jiǎn)化后的代碼：

private Runnable getTask() {if (一些特殊情況){return null;}Runnable r = workQueue.take();return r; }

這個(gè)workQueue就是初始化ThreadPoolExecutor時(shí)存放任務(wù)的BlockingQueue隊(duì)列，這個(gè)隊(duì)列里的存放的都是將要執(zhí)行的Runnable任務(wù)。因?yàn)锽lockingQueue是個(gè)阻塞隊(duì)列，BlockingQueue.take()得到如果是空，則進(jìn)入等待狀態(tài)直到BlockingQueue有新的對(duì)象被加入時(shí)喚醒阻塞的線程。所以一般情況Thread的run()方法就不會(huì)結(jié)束,而是不斷執(zhí)行從workQueue里的Runnable任務(wù)，這就達(dá)到了線程復(fù)用的原理了。

2.控制最大并發(fā)數(shù)

那Runnable是什么時(shí)候放入workQueue？Worker又是什么時(shí)候創(chuàng)建，Worker里的Thread的又是什么時(shí)候調(diào)用start()開(kāi)啟新線程來(lái)執(zhí)行Worker的run()方法的呢？有上面的分析看出Worker里的runWorker()執(zhí)行任務(wù)時(shí)是一個(gè)接一個(gè)，串行進(jìn)行的，那并發(fā)是怎么體現(xiàn)的呢？

很容易想到是在execute(Runnable runnable)時(shí)會(huì)做上面的一些任務(wù)。看下execute里是怎么做的。

execute:

簡(jiǎn)化后的代碼

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();//當(dāng)前線程數(shù) < corePoolSizeif (workerCountOf(c) < corePoolSize) {//直接啟動(dòng)新的線程if (addWorker(command, true))return;c = ctl.get();}// 活動(dòng)線程數(shù) >= corePoolSize// runState為RUNNING && 隊(duì)列未滿if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();//再次檢驗(yàn)是否為RUNNING狀態(tài)//非RUNNING狀態(tài),則從workQueue中移除任務(wù)并拒絕if (!isRunning(recheck) && remove(command)){reject(command);//采用線程池指定的策略拒絕任務(wù) // 兩種情況：// 1.非RUNNING狀態(tài)拒絕新的任務(wù)// 2.隊(duì)列滿了啟動(dòng)新的線程失敗（workCount > maximumPoolSize }else if (workerCountOf(recheck) == 0){addWorker(null, false);}}else if (!addWorker(command, false))reject(command); }

addWorker:

簡(jiǎn)化后的代碼

private boolean addWorker(Runnable firstTask, boolean core) {int wc = workerCountOf(c);if (wc >= (core ? corePoolSize : maximumPoolSize)){return false;}Worker w = new Worker(firstTask);final Thread t = w.thread;t.start(); }

根據(jù)代碼再來(lái)看上面提到的線程池工作過(guò)程中的添加任務(wù)的情況：

* 如果正在運(yùn)行的線程數(shù)量小于 corePoolSize，那么馬上創(chuàng)建線程運(yùn)行這個(gè)任務(wù)；
* 如果正在運(yùn)行的線程數(shù)量大于或等于corePoolSize，那么將這個(gè)任務(wù)放入隊(duì)列；
* 如果這時(shí)候隊(duì)列滿了，而且正在運(yùn)行的線程數(shù)量小于 maximumPoolSize，那么還是要?jiǎng)?chuàng)建非核心線程立刻運(yùn)行這個(gè)任務(wù)；
* 如果隊(duì)列滿了，而且正在運(yùn)行的線程數(shù)量大于或等于 maximumPoolSize，那么線程池會(huì)拋出異常RejectExecutionException。

通過(guò)addWorker如果成功創(chuàng)建新的線程成功，則通過(guò)start()開(kāi)啟新線程，同時(shí)將firstTask作為這個(gè)Worker里的run()中執(zhí)行的第一個(gè)任務(wù)。

雖然每個(gè)Worker的任務(wù)是串行處理，但如果創(chuàng)建了多個(gè)Worker，因?yàn)楣灿靡粋€(gè)workQueue，所以就會(huì)并行處理了。

所以根據(jù)corePoolSize和maximumPoolSize來(lái)控制最大并發(fā)數(shù)。大致過(guò)程可用下圖表示。

上面的講解和圖來(lái)可以很好的理解的這個(gè)過(guò)程。

3.管理線程

通過(guò)線程池可以很好的管理線程的復(fù)用，控制并發(fā)數(shù)，以及銷毀等過(guò)程,線程的復(fù)用和控制并發(fā)上面已經(jīng)講了，而線程的管理過(guò)程已經(jīng)穿插在其中了，也很好理解。

在ThreadPoolExecutor有個(gè)ctl的AtomicInteger變量。通過(guò)這一個(gè)變量保存了兩個(gè)內(nèi)容：

所有線程的數(shù)量、每個(gè)線程所處的狀態(tài)，其中低29位存線程數(shù)，高3位存runState，通過(guò)位運(yùn)算來(lái)得到不同的值。

private final AtomicInteger ctl = newAtomicInteger(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; }//判斷線程是否在運(yùn)行 private static boolean isRunning(int c) {return c < SHUTDOWN; }

這里主要通過(guò)shutdown和shutdownNow()來(lái)分析線程池的關(guān)閉過(guò)程。首先線程池有五種狀態(tài)來(lái)控制任務(wù)添加與執(zhí)行。主要介紹以下三種：

RUNNING狀態(tài)：線程池正常運(yùn)行，可以接受新的任務(wù)并處理隊(duì)列中的任務(wù)；

SHUTDOWN狀態(tài)：不再接受新的任務(wù)，但是會(huì)執(zhí)行隊(duì)列中的任務(wù)；

STOP狀態(tài)：不再接受新任務(wù)，不處理隊(duì)列中的任務(wù)。

shutdown這個(gè)方法會(huì)將runState置為SHUTDOWN，會(huì)終止所有空閑的線程，而仍在工作的線程不受影響，所以隊(duì)列中的任務(wù)仍會(huì)被執(zhí)行。

shutdownNow方法將runState置為STOP。和shutdown方法的區(qū)別，這個(gè)方法會(huì)終止所有的線程，所以隊(duì)列中的任務(wù)也不會(huì)被執(zhí)行了。

總結(jié)

通過(guò)對(duì)ThreadPoolExecutor源碼的分析，從總體上了解了線程池的創(chuàng)建，任務(wù)的添加，執(zhí)行等過(guò)程，熟悉這些過(guò)程，使用線程池就會(huì)更輕松了。

而從中學(xué)到的一些對(duì)并發(fā)控制，以及生產(chǎn)者——消費(fèi)者模型任務(wù)處理的使用，對(duì)以后理解或解決其他相關(guān)問(wèn)題會(huì)有很大的幫助。

總結(jié)

以上是生活随笔為你收集整理的Java线程池，从使用到原理的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。