這是一篇 寫的非常用心的 博文,這里是原文地址
https://blog.csdn.net/l540675759/article/details/62230562
前言
(1)本文共花費(fèi)2周零3天的凌晨時光,這段時間收獲很多.
(2)從整理文章,作者從線程-->阻塞隊列-->二進(jìn)制-->線程池的內(nèi)部機(jī)制,一路走來,本來是想寫一篇為AsyncTask做鋪墊的文章,沒想到越寫越多.
(3)文章中如果錯誤,請大家及時指正,作者會及時更新.
(4)希望大家能夠從文章中.多多收獲,迄今為止,博主最好的一篇文章,也是花了大力氣最用心的一篇文章.
線程
在了解線程池之前，先給大家介紹下線程的概念：
先看一個燒水的例子，圖中看電視是主線，用戶想在看電視的過程中去完成燒水這個操作，并且不耽誤看電視，看了這張圖，在去了解接下來的概念會更好的理解主線程與子線程的概念。
線程是什么？
從底層角度來說：
一個線程就是在進(jìn)程中的一個單一的順序控制流.而單個進(jìn)程可以擁有多個并發(fā)執(zhí)行的任務(wù)，每個任務(wù)都好像有自己的CPU一樣，而其底層的機(jī)制就是切分CPU的時間，也就是CPU將輪流給每個任務(wù)分配其占用時間。
每個任務(wù)都覺得自己在一直占用CPU，而事實上是將CPU時間劃分成片段分配給所有的任務(wù)。
在多個CPU的環(huán)境下，多線程的運(yùn)作，可以極大的提供程序的運(yùn)行速度，這就是線程存在的意義。
那么在Android中，線程的作用是？
首先，先了解下Android下進(jìn)程和線程的概念：
這里引用Gityuan作者在知乎上的回答，關(guān)于線程和進(jìn)程的概念
進(jìn)程：每個app運(yùn)行時前首先創(chuàng)建一個進(jìn)程，該進(jìn)程是由Zygote fork出來的，用于承載App上運(yùn)行的各種Activity/Service等組件。
進(jìn)程對于上層應(yīng)用來說是完全透明的，這也是google有意為之，讓App程序都是運(yùn)行在Android Runtime。大多數(shù)情況一個App就運(yùn)行在一個進(jìn)程中，除非在AndroidManifest.xml中配置Android:process屬性，或通過native代碼fork進(jìn)程。線程：線程對應(yīng)用來說非常常見，比如每次new Thread().start都會創(chuàng)建一個新的線程。該線程與App所在進(jìn)程之間資源共享，從Linux角度來說進(jìn)程與線程除了是否共享資源外，并沒有本質(zhì)的區(qū)別，都是一個task_struct結(jié)構(gòu)體，在CPU看來進(jìn)程或線程無非就是一段可執(zhí)行的代碼，CPU采用CFS調(diào)度算法，保證每個task都盡可能公平的享有CPU時間片。
上面可能還是比較專業(yè)，這里簡要總結(jié)下線程在Android的作用：
（1）在Android中線程分主線程和子線程，主線程也被稱為UI線程，用來處理各種和界面相關(guān)的事情，
例 ：界面的加載，Activity的生命周期這些都在主線程的范疇之內(nèi)。
（2）由于主線程比較特殊，因為本身主線程在處理界面上，用了大部分的消耗，所以主線程不能再處理過于耗時的操作（IO操作，網(wǎng)絡(luò)請求，大量的數(shù)據(jù)操作），否則就會造成ANR現(xiàn)象（程序卡死）。
而造成這種現(xiàn)象的主要原因有：
Activity響應(yīng)時間超過5s
Broadcast在處理時間超過10s
Service處理時間超過20s
這大部分的原因是主線程進(jìn)行過于耗時的操作，因為Activity，Broadcast，Serivce本身都是通過主線程進(jìn)行承載的。
（3）此時子線程就橫空出世解決了這類問題，Android建議耗時操作必須放在子線程中運(yùn)行。
（4）而在Android中可以解決耗時問題的角色除了Thread之外還有AsyncTask，HandlerThread，IntentService，都可以實現(xiàn)此類功能，而他們的本質(zhì)還是傳統(tǒng)的線程。
為什么會有線程池？
從字面上來看，線程池是存放，和管理線程的池子。那么為什么會有線程池呢？
先看一個例子，這里我用Handler和Thread來模擬網(wǎng)絡(luò)請求的操作：
private Handler mHandler = new Handler(new Handler.Callback() {
@Override
public boolean handleMessage(Message msg) {
if (msg.what == TASK_ACTION) {
Log.d("收到消息", "更新UI");
}
return false;
}
});
new Thread(new Runnable() {
@Override
public void run() {
try {
//模擬網(wǎng)絡(luò)請求
Thread.sleep(1000);
mHandler.sendEmptyMessage(TASK_ACTION);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
上面過程，只是用一個Thread來模擬正常的網(wǎng)絡(luò)請求，然后通過Handler來回調(diào)給UI線程，通知UI線程來刷新，如果對Handler機(jī)制不太了解，
上面只是單純的一個網(wǎng)絡(luò)請求，那么現(xiàn)在需求來了，這個界面不止一個網(wǎng)絡(luò)請求，可能存在大量的網(wǎng)絡(luò)請求，這時候就會有問題產(chǎn)生：
（1）當(dāng)大量的網(wǎng)絡(luò)請求產(chǎn)生，就會大量的創(chuàng)建和銷毀線程，因此可能會造成過大的性能開銷。
（2）當(dāng)大量的線程一起運(yùn)作的時候，可能會造成資源緊張，上面也介紹過線程底層的機(jī)制就是切分CPU的時間，而大量的線程同時存在時可能造成互相搶占資源的現(xiàn)象發(fā)生，從而導(dǎo)致阻塞的現(xiàn)象。
基于以上背景，線程池適當(dāng)?shù)某霈F(xiàn)可以很好的解決上述的問題，而上述模擬網(wǎng)絡(luò)請求也只是一個簡單的例子，而現(xiàn)實情況下，會有好多種情況和上述相似，比如在數(shù)據(jù)庫操作大數(shù)據(jù)，多線程下載，在使用Thread的同時都會出現(xiàn)上述情況。
什么是線程池？
Android中的線程池的概念來源于Java中的Executor，Executor是一個接口，真正的線程池的實現(xiàn)為ThreadPoolExecutor，ThreadPoolExecutor提供了一系列參數(shù)來配置線程池，通過不同的參數(shù)可以創(chuàng)建不同的線程池。
線程池的優(yōu)點：
線程池的出現(xiàn)，恰恰就是解決上面類似問題的痛點，而線程池的優(yōu)點有：
（1）復(fù)用線程池中的線程，避免因為線程的創(chuàng)建和銷毀所帶來的性能開銷。
（2）能夠有效的控制線程池的最大并發(fā)數(shù)，避免大量的線程之間因互相搶占系統(tǒng)資源而導(dǎo)致的阻塞現(xiàn)象。
（3）能夠?qū)€程進(jìn)行簡單的管理，并提供定時執(zhí)行以及指定間隔循環(huán)執(zhí)行等功能。
線程池的構(gòu)造方法
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory)
上面代碼是創(chuàng)建一個基本的線程池需要的參數(shù)，讓我們通過圖來簡要的描述下：
由上圖可以簡要的描述出創(chuàng)建一個基本的線程池需要的參數(shù)，以及各個參數(shù)的含義，下面將詳細(xì)說明各個參數(shù)的具體含義。
CorePoolSize
線程的核心線程數(shù)。
默認(rèn)情況下，核心線程數(shù)會在線程中一直存活，即使它們處于閑置狀態(tài)。
如果將ThreadPoolExecutor的allowCoreThreadTimeOut屬性設(shè)置為true，那么核心線程就會存在超時策略，這個時間間隔有keepAliveTime所決定，當(dāng)?shù)却龝r間超過keepAliveTime所指定的時長后，核心線程就會被停止。
maximumPoolSize
線程池所能容納的最大線程數(shù)。
當(dāng)活動線程數(shù)達(dá)到這個數(shù)值后，后續(xù)的新任務(wù)將會被阻塞。
keepAliveTime
非核心線程閑置時的超時時長，超過這個時長，非核心線程就會被回收，當(dāng)ThreadPoolExector的allowCoreThreadTimeOut屬性設(shè)置為True時，keepAliveTime同樣會作用于核心線程。
unit
用于指定keepAliveTime參數(shù)的時間單位，這是一個枚舉，常用的有TimeUnit.MILLISECONDS（毫秒）、TimeUnit.SECONDS(秒)以及TimeUnit.MINUTES(分鐘)等。
TimeUnit.NANOSECONDS 納秒
TimeUnit.MICROSECONDS 微秒
TimeUnit.MILLISECONDS 毫秒
TimeUnit.SECONDS 秒
TimeUnit.MINUTES 分鐘
TimeUnit.HOURS 小時
TimeUnit.DAYS 天
workQueue
線程池中的任務(wù)隊列，通過線程池execute方法提交的Runnable對象會存儲在這個參數(shù)中。
這個任務(wù)隊列是BlockQueue類型，屬于阻塞隊列，就是當(dāng)隊列為空的時候，此時取出任務(wù)的操作會被阻塞，等待任務(wù)加入隊列中不為空的時候，才能進(jìn)行取出操作，而在滿隊列的時候，添加操作同樣被阻塞。
如果有想了解的可以參考下這篇文章：
Java多線程-工具篇-BlockingQueue
threadFactory
線程工廠，為線程池提供創(chuàng)建新線程的功能。ThreadFactory是一個接口，它只有一個方法，newThread（Runnable r），用來創(chuàng)建線程。
ThreadFactory factory =new ThreadFactory() {
//線程安全的Integer操作類
private final AtomicInteger mCount =new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "new Thread #" + mCount.getAndIncrement());
}
};
線程池的源碼解析
打開源碼,先把線程池源碼中除了構(gòu)造參數(shù),其他的一些基本屬性,先給分析一下.
線程池的生命周期
//這里在線程池統(tǒng)計數(shù)值,用AtomicInteger,它是一種線程安全的加減操作類
//初始生命周期是RUNNING,工作線程的初始數(shù)量是0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//進(jìn)行移位操作需要的常量 Integer.SIZE =32 bit位
private static final int COUNT_BITS = Integer.SIZE - 3;
//進(jìn)行位運(yùn)算需要的常量
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//進(jìn)行高位運(yùn)算
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;
將上述高位運(yùn)算就是將 0 和1以及其他的數(shù)值在二進(jìn)制下,向左移位29位,缺位用0補(bǔ)齊,實際結(jié)果就變成:
# 接受新任務(wù),并且處理隊列任務(wù)的狀態(tài)
RUNNING = 111 000...000 (29個0)
# 不接受新任務(wù),但是會處理隊列任務(wù)的狀態(tài)
SHUTDOWN = 000 000...000 (29個0不包括前三位)
# 不接受新任務(wù),并且也不會處理隊列任務(wù)的狀態(tài)
STOP = 001 000...000 (29個0)
# 所有線程池內(nèi)線程都將被終止,并且將workCount清零,在這里狀態(tài)下將會運(yùn)行terminated()方法(終止線程池的方法)
TIDYING = 010 000...000 (29個0)
# terminated()方法以及結(jié)束的狀態(tài)
TERMINATED = 011 000...000 (29個0)
/**
* 獲取到當(dāng)前線程池的生命周期的狀態(tài)
*/
private static int runStateOf(int c) { return c & ~CAPACITY; }
/**
* 獲取當(dāng)前線程池的工作線程狀態(tài)
*/
private static int workerCountOf(int c) { return c & CAPACITY; }
/**
* 通過或運(yùn)算拼接線程的生命周期狀態(tài)和工作線程的個數(shù)
*/
private static int ctlOf(int rs, int wc) { return rs | wc; }
上面的三個函數(shù)是獲取當(dāng)前線程池狀態(tài)的方法,這里簡單介紹下:
(1) ctlOf()有兩個參數(shù),一個是生命周期狀態(tài),一個是當(dāng)前線程池工作線程.
生命周期的狀態(tài)格式:
XXX 0000…0000(29個0)
ctlOf()返回的值就是將工作線程數(shù)量轉(zhuǎn)化成2進(jìn)制拼接在生命周期的二進(jìn)制后半段上.
(2) runStateOf()和workerCountOf()方法都是讓生命周期的狀態(tài)值與CAPACITY和CAPACITY的反碼進(jìn)行與運(yùn)算,簡明的說,就是獲得二進(jìn)制數(shù)的高位(前三位)和低位(后29位).
如果大家比較了解位運(yùn)算可以發(fā)現(xiàn):
CAPACITY ------> 000 1111...1111 (29個1)
~CAPACITY ------> 111 0000...0000 (29個0)
所以在進(jìn)行與運(yùn)算的同時,可以分別取出前3位和后29位,來分別代表線程池的生命周期和工作線程數(shù).
其他屬性
/**
* 無法執(zhí)行任務(wù)的通知類
* 在Android中不太常用
*/
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
當(dāng)線程池?zé)o法執(zhí)行任務(wù),這可能由于任務(wù)隊列已滿或者是無法成功執(zhí)行任務(wù).這個時候ThreadPoolExecution就會調(diào)用handler的rejectedExecution方法來通知調(diào)用者.
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
默認(rèn)情況下,rejectedExecution會拋出個RejectedExecutionException異常,來說明為什么當(dāng)前無法執(zhí)行任務(wù).
ThreadPoolExecution為RejectedExecutionException提供了幾個可選值:
----------------------------CallerRunsPolicy-------------------
//拒絕任務(wù)時,判斷線程池的狀態(tài)是否為SHUTDOWN,如果是任務(wù)將會被丟棄,如果不是的話任務(wù)會被繼續(xù)執(zhí)行.
public void rejectedExecution(Runnable r, ThreadPoolExecutor e){
if (!e.isShutdown()) {
r.run();
}
}
-------------------------AbortPolicy(默認(rèn)值)---------------------
//拒絕任務(wù)時,直接拋出異常和原因
public void rejectedExecution(Runnable r, ThreadPoolExecutor e){
throw new RejectedExecutionException(
"Task " + r.toString() +
" rejected from " +e.toString());
}
-------------------------DiscardPolicy--------------------------
//就是單純的拒絕任務(wù)而已,什么也不會發(fā)生,任務(wù)也將丟失public void rejectedExecution(Runnable r, ThreadPoolExecutor e){
//什么沒發(fā)生
}
----------------------DiscardOldestPolicy-----------------------
//拒絕任務(wù)時,判斷線程池的狀態(tài)是否為SHUTDOWN,如果是任務(wù)將會被丟棄,如果不是的話,將當(dāng)前請求隊列中等待時間最長的任務(wù)彈出,將其加入隊列中.
public void rejectedExecution(Runnable r, ThreadPoolExecutor e){
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
比較重要的方法
線程池有兩個執(zhí)行的方法,分別是submit()和execute(),這兩個方法本質(zhì)的含義是一樣的.
從圖上可以看出的,submit()其實還是需要調(diào)用execute()去執(zhí)行任務(wù),而submit()和execute()本質(zhì)上的不同是submit()將包裝好的任務(wù)進(jìn)行了返回.
submit()
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
//還是通過調(diào)用execute
execute(ftask);
//最后會將包裝好的Runable返回
return ftask;
}
//將Callable<T> 包裝進(jìn)FutureTask中
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
//可以看出FutureTask也是實現(xiàn)Runnable接口,因為RunableFuture本身就繼承了Runnabel接口
public class FutureTask<V> implements RunnableFuture<V> {
.......
}
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
execute()
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//獲得當(dāng)前線程的生命周期對應(yīng)的二進(jìn)制狀態(tài)碼
int c = ctl.get();
//判斷當(dāng)前線程數(shù)量是否小于核心線程數(shù)量,如果小于就直接創(chuàng)建核心線程執(zhí)行任務(wù),創(chuàng)建成功直接跳出,失敗則接著往下走.
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//判斷線程池是否為RUNNING狀態(tài),并且將任務(wù)添加至隊列中.
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//審核下線程池的狀態(tài),如果不是RUNNING狀態(tài),直接移除隊列中
if (! isRunning(recheck) && remove(command))
reject(command);
//如果當(dāng)前線程數(shù)量為0,則單獨創(chuàng)建線程,而不指定任務(wù).
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果不滿足上述條件,嘗試創(chuàng)建一個非核心線程來執(zhí)行任務(wù),如果創(chuàng)建失敗,調(diào)用reject()方法.
else if (!addWorker(command, false))
reject(command);
}
下圖是一張execute()方法的基本流程:
從execute()方法中,能看出addWorker()方法,是創(chuàng)建線程(核心線程,非核心線程)的主要方法,而reject()就是線程創(chuàng)建失敗的一個回調(diào).
reject()
那我們來看一下reject()方法,這里就是通過上述的Handler將通知發(fā)出去.然后針對不同的類型的RejectedExecutionHandler,進(jìn)行不同的處理,這里我們上文有介紹.
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
下面我們著重看下創(chuàng)建線程的方法:
addWorker()
參數(shù) :
Runnable firstTask:
為傳遞進(jìn)來需要執(zhí)行的任務(wù),也可以設(shè)置為null(在SHUTDOWN情況下,單純的創(chuàng)建線程來執(zhí)行任務(wù)).
boolean core:
需要創(chuàng)建的線程是否需要是核心線程.
private boolean addWorker(Runnable firstTask, boolean core) {
//類似goto,是Java的標(biāo)識符,在這里出現(xiàn)是為了防止在多線程的情況下,compareAndIncrementWorkerCount(),計算線程池狀態(tài)出現(xiàn)問題,而設(shè)立重試的關(guān)鍵字.
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//看似判斷條件很麻煩
//分拆后主要兩點
//線程已經(jīng)處于STOP或者即將STOP的狀態(tài)
//或者 處于SHUTDOWN狀態(tài),并且傳遞的任務(wù)為null,此時隊列不為空還需要增加線程,除了這種情況,其他情況都不需要增加線程
//以上的情況就不需要
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//判斷當(dāng)前工作線程數(shù)量是否超過最大值
//或者當(dāng)前工作線程數(shù)量超過 核心線程數(shù)或者最大線程數(shù),這個值根據(jù)第二個布爾變量決定
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//這段函數(shù)是判斷 線程池狀態(tài)的統(tǒng)計更新成沒成功
//如果成功直接跳出這個循環(huán),繼續(xù)執(zhí)行
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
//如果不成功則跳到外層循環(huán)入口,重新執(zhí)行.
retry inner loop
}
}
//下面是創(chuàng)建線程的過程,并且在創(chuàng)建線程的過程中加鎖
//Worker就是線程的一個包裝類.
//這里分別對線程的創(chuàng)建成功和失敗分別做出了處理.
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//創(chuàng)建線程的過程中,加鎖防止并發(fā)現(xiàn)象發(fā)生.
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
//從這里可以看出線程池創(chuàng)建線程,只會在兩種情況下創(chuàng)建:
//1.線程池在RUNNING狀態(tài)(rs<SHUTDOWN)
//2.線程池處于SHUTDOWN狀態(tài),并且任務(wù)為null,但是此時任務(wù)隊列不為空,需要繼續(xù)增加線程來加快處理進(jìn)度.
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//在這里就是先檢查下Thread狀態(tài),防止意外發(fā)生.
if (t.isAlive())
throw new IllegalThreadStateException();
workers.add(w);
//這里做了一個容量的判斷
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//如果線程已經(jīng)增加成功,然后設(shè)置標(biāo)志
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//最后如果線程沒有開始,就分發(fā)到添加線程失敗,通過標(biāo)志位來判斷線程是否被添加成功.
if (! workerStarted)
addWorkerFailed(w);
}
//如果添加成功就返回true,否則添加失敗就返回false.
return workerStarted;
}
addWorker()方法的注意事項:
(1)增加一個線程,并且會為其綁定core或者maximum的線程標(biāo)志.
(2)如果成功添加線程來執(zhí)行當(dāng)前任務(wù),那么當(dāng)前線程池的狀態(tài)會被刷新.
(3)在添加第一個任務(wù)firstTask的這種情況下,新的工作線程會被創(chuàng)建后立即執(zhí)行任務(wù).
(4)該方法會在線程池STOP狀態(tài)或者符合資格去關(guān)閉會返回false.
(5)線程工廠創(chuàng)建線程失敗的時候,同樣也會返回false.
(6)在由于線程創(chuàng)建失敗,線程工廠返回的線程為null,或者發(fā)生異常(通常由于在線程執(zhí)行的過程中發(fā)生了OOM),線程池會進(jìn)行回滾操作.
addWorker()方法執(zhí)行的幾個階段
第一階段 :
狀態(tài)檢查
在創(chuàng)建線程時,首先檢查線程池狀態(tài),防止線程處于STOP,TIDYING,TERMINATED狀態(tài),如果處于上述狀態(tài)直接返回false.
然后對于在SHUTDOWN狀態(tài)下,只有當(dāng)前任務(wù)隊列不為空,并且傳遞的任務(wù)參數(shù)為null.這種狀態(tài)下可以創(chuàng)建線程來執(zhí)行剩余任務(wù),除此之外全部直接返回false.
if (rs >= SHUTDOWN &&! (rs == SHUTDOWN && firstTask == null
&&! workQueue.isEmpty()))
return false;
第二階段 :
判斷當(dāng)前線程池的能否創(chuàng)建線程以及可以創(chuàng)建之后的數(shù)量添加校驗
(1)當(dāng)前線程的數(shù)量是否超過線程池的最大容量,以及根據(jù)core參數(shù)來判斷是否超過設(shè)置的核心線程數(shù),和最大線程數(shù).
(2)通過第一步之后就可以創(chuàng)建線程,這里需要用到compareAndIncrementWorkerCount()通過原子操作來更新線程池的線程數(shù)量變化,如果變化數(shù)量失敗,這里有一個重試機(jī)制,這個retry關(guān)鍵字就是來完成這個操作.
(3)這里注明下CAPACITY這個常量就是線程池的線程數(shù)量的極限
CAPACITY = 1>>29 -1 =2^29-1
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
}
第三階段 :
創(chuàng)建線程
通過上述階段,那么就可以創(chuàng)建線程了,這里設(shè)置了兩個初始的標(biāo)志位,來判斷被創(chuàng)建線程的狀態(tài).
boolean workerStarted = false;
boolean workerAdded = false;
如果最終線程創(chuàng)建并添加成功,則返回true,如果線程最終沒有被運(yùn)行,則調(diào)用addWorkerFailed()方法.
由于邏輯并不復(fù)雜,這里就不貼代碼了.
其他相關(guān)方法
addWorkedFailed()
在addWorker()方法中,如果線程創(chuàng)建之后,沒有最終運(yùn)行(workerStarted=false)這時候會調(diào)用addWorkedFailed()方法.
/**
* 回滾工作線程的創(chuàng)建操作:
* 1.如果線程的包裝類Worker存在,就將其remove掉.
* 2.remove掉添加線程失敗的Worker,需要刷新當(dāng)前工作線程的數(shù)量
* 3.嘗試終止操作,并且終止這個線程的操作.
*/
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
//嘗試停止操作.
tryTerminate();
} finally {
mainLock.unlock();
}
}
tryTerminate()
而在addWorkedFailed()方法中,我們發(fā)現(xiàn)除了回滾操作,它還調(diào)用了tryTerminate()方法,嘗試著去停止線程池.因為線程池創(chuàng)建線程失敗一般由于異常引起(或OOM),所以這時候需要讓線程池進(jìn)行停止操作.
注意事項:
如果發(fā)生以下兩種情況,使用該方法將會將線程池轉(zhuǎn)換為終止?fàn)顟B(tài)(TERMINATED):
1.SHUTDOWN狀態(tài)下,隊列為空的情況下.
2.STOP狀態(tài)下.
如果符合上述條件,可以轉(zhuǎn)換終止?fàn)顟B(tài)時,這時會中斷當(dāng)前線程池內(nèi)空閑的線程,以確保終止的信號的傳遞.
final void tryTerminate() {
for (;;) {
int c = ctl.get();
//檢測當(dāng)前是RUNNING狀態(tài),或者已經(jīng)停止(TERMINATED)的狀態(tài),或者SHUTDOWN狀態(tài)下,隊列不為空.
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//如果工作線程的數(shù)量不為空,這時候需要處理空閑線程,這里只中斷一個其中一個線程,這里博主認(rèn)為是將線程池的狀態(tài)由SHUTDOWN向STOP狀態(tài)過渡的信號.
if (workerCountOf(c) != 0) {
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//設(shè)置當(dāng)前的線程池狀態(tài)為TIDYING,如果設(shè)置失敗,還會進(jìn)入循環(huán)直到設(shè)置成功.
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
//停止方法的空實現(xiàn)
terminated();
} finally {
//最終線程池會設(shè)置為停止?fàn)顟B(tài)
ctl.set(ctlOf(TERMINATED, 0));
//設(shè)置可重新入鎖的標(biāo)志,將被鎖隔離的在外等待的所有線程喚醒.
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
}
}
interruptIdleWorkers()
而在tryTerminate()方法中,這里中斷線程的操作就是由interruptIdleWorkers()方法進(jìn)行的.
這個方法作用很明確,就是設(shè)置線程中斷操作的方法,唯一注意的地方就是參數(shù)onlyOne:
如果為true,只中斷工作線程中的一個線程.
如果為false,中斷所有的工作線程.
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
//檢查線程的狀態(tài)
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
//如果onlyOne參數(shù)為True,則只執(zhí)行一次就跳出.
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
shutdown()
而中斷所有空閑的線程方法則是shutdown()方法,它的核心方法還是調(diào)用interruptIdleWorkers()方法.
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//校驗線程的狀態(tài)
checkShutdownAccess();
//設(shè)置線程池狀態(tài)為SHUTDOWN
advanceRunState(SHUTDOWN);
//中斷所有空閑進(jìn)程.調(diào)用的interruptIdleWorkers(false);
interruptIdleWorkers();
//需要自己實現(xiàn),在中斷所有線程可定制的操作
onShutdown();
} finally {
mainLock.unlock();
}
tryTerminate();
}
注意事項:
(1)在shutdown()執(zhí)行時可以讓現(xiàn)有的任務(wù)被執(zhí)行,但是新的任務(wù)不在會被處理.
(2)如果已經(jīng)是SHUTDOWN狀態(tài),那么繼續(xù)調(diào)用不會產(chǎn)生任何效果.
(3)shutdown()方法只會中斷空閑的線程,但是不會影響到已經(jīng)存入隊列的任務(wù),如果需要停止線程池的運(yùn)行,可以使用awaitTermination()方法.
awaitTermination()
阻塞方法,強(qiáng)行等待當(dāng)前隊列中的任務(wù)全部為TERMINATED狀態(tài),可以設(shè)置超時時間.
參數(shù):d
timeout —- 設(shè)置超時時間
unit —- 設(shè)置超時時間的單位
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
//設(shè)置時間
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//這是死循環(huán),當(dāng)線程池的狀態(tài)為TERMINATED時,跳出循環(huán)返回true,也就是所有任務(wù)都完成.否則超時或者線程中斷則返回false.
while (!runStateAtLeast(ctl.get(), TERMINATED)) {
if (nanos <= 0L)
return false;
nanos = termination.awaitNanos(nanos);
}
return true;
} finally {
mainLock.unlock();
}
}
線程池的分類
Android中最常見的四類具有不同功能特性的線程池:
1.FixedThreadPool
//特點:
//核心線程數(shù)和最大線程數(shù)相同.
//無超時時間
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(
nThreads, nThreads,
0L, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>()
);
這是一種數(shù)量固定的線程池,當(dāng)線程處于空閑的時候,并不會被回收,除非線程池被關(guān)閉.
當(dāng)所有的線程都處于活動狀態(tài)時,新任務(wù)都會處于等待狀態(tài),直到有線程空閑出來.
由于FixedThreadPool中只有核心線程并且這些核心線程不會被回收,這意味著它能夠更加快速地響應(yīng)外界的請求.
通過構(gòu)造方法可以看出,FixedThreadPool只有核心線程,并且超時時間為0(即無超時時間),所以不會被回收.
2.CacheThreadPool
//無核心線程,并且最大線程數(shù)為int的最大值.
//超時時間為60s
//隊列為SynchronousQueue同步阻塞隊列,隊列中沒有任何容量.只有在有需求的情況下,隊列中才可以試著添加任務(wù).
public static ExecutorService newCacheThreadPool(){
return new ThreadPoolExecutor(
0,Integer.MAX_VALUE,
60L,TimeUnit.SECONDS,
new SynchronousQueue<Runnable>()
);
}
它是一種線程數(shù)量不定的線程池,它只有非核心線程,并且其最大線程數(shù)為Integer.MAX_VALUE(也就相當(dāng)于線程池的線程數(shù)量可以無限大).
當(dāng)線程池中所有線程都處于活動的狀態(tài)時,線程池會創(chuàng)建新的線程來處理新任務(wù),否則就會復(fù)用空閑線程來處理.
值得注意的是,這個線程池中儲存任務(wù)的隊列是SynchronousQueue隊列,這個隊列可以理解為無法儲存的隊列,只有在可以取出的情況下,才會向其內(nèi)添加任務(wù).
從整個CacheThreadPool的特性來看:
(1)比較適合執(zhí)行大量的耗時較少的任務(wù).
(2)當(dāng)整個線程都處于閑置狀態(tài)時,線程池中的線程都會超時而被停止,這時候的CacheThreadPool幾乎不占任何系統(tǒng)資源的.
3.ScheduledThreadPool
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSzie) {
return new ScheduledThreadPoolExecutor(corePoolSzie);
}
//核心線程數(shù)是固定的,非核心線程無限大,并且非核心線程數(shù)有10s的空閑存活時間
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
new DelayedWorkQueue());
}
它的核心線程數(shù)量是固定的,而非核心線程數(shù)是沒有限制的,并且當(dāng)非核心線程閑置時會被立即回收.
ScheduThreadPool這類線程池主要用于執(zhí)行定時任務(wù)和具有固定周期的重復(fù)任務(wù).
而DelayedWorkQueue這個隊列就是包裝過的DelayedQueue,這個類的特點是在存入時會有一個Delay對象一起存入,代表需要過多少時間才能取出,相當(dāng)于一個延時隊列.
4.SingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
return Executors.newSingleThreadExecutor();
}
//特點:
//線程中只有一個核心線程
//并且無超時時間
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
這類線程池內(nèi)部只有一個核心線程,它確保所有的任務(wù)都在同一個線程中按順序執(zhí)行.
SingleThreadExecutor的意義在于統(tǒng)一外界所有任務(wù)到一個線程,這使得這些任務(wù)之間不需要處理線程同步的問題.
參考文檔:
1.安卓開發(fā)藝術(shù)探索
2.ThreadPoolExecutor解析-主要源碼研究
http://blog.csdn.net/wenhuayuzhihui/article/details/51377174
3.理解java線程的中斷(interrupt)
http://blog.csdn.net/canot/article/details/51087772
</article>
