并發(fā)編程—線程池核心原理分析
作者：一起擼Java
線程執(zhí)行完run發(fā)放自動被銷毀了，且任務(wù)與線程綁定在了一起，所以當(dāng)任務(wù)多的時候，會頻繁的創(chuàng)建和銷毀線程，這給我們CPU和內(nèi)存帶來了很大的開銷。
第1章 線程池簡介
1、線程的問題
- 線程執(zhí)行完run發(fā)放自動被銷毀了，且任務(wù)與線程綁定在了一起，所以當(dāng)任務(wù)多的時候，會頻繁的創(chuàng)建和銷毀線程，這給我們CPU和內(nèi)存帶來了很大的開銷。
- 線程一多了，無法實現(xiàn)統(tǒng)一管理。
2、線程池的概念及作用
- 他是池化技術(shù)的一種應(yīng)用
- 他實現(xiàn)了線程的重復(fù)利用
- 實現(xiàn)了對線程資源的管理控制
3、常見線程池
- newFixedThreadPool：該方法返回一個固定數(shù)量的線程池，線程數(shù)不變，當(dāng)有一個任務(wù)提交時，若線程池中空閑，則立即執(zhí)行，若沒有，則會被暫緩在一個任務(wù)隊列中，等待有空閑的線程去執(zhí)行。
- newSingleThreadExecutor: 創(chuàng)建一個線程的線程池，若空閑則執(zhí)行，若沒有空閑線程則暫緩在任務(wù)隊列中。
- newCachedThreadPool：返回一個可根據(jù)實際情況調(diào)整線程個數(shù)的線程池，不限制最大線程數(shù)量，若用空閑的線程則執(zhí)行任務(wù)，若無任務(wù)則不創(chuàng)建線程。并且每一個空閑線程會在60秒后自動回收
- newScheduledThreadPool: 創(chuàng)建一個可以指定線程的數(shù)量的線程池，但是這個線程池還帶有延遲和周期性執(zhí)行任務(wù)的功能，類似定時器。
- newWorkStealingPool:適合使用在很耗時的操作，但是newWorkStealingPool不是ThreadPoolExecutor的擴(kuò)展，它是新的線程池類ForkJoinPool的擴(kuò)展，但是都是在統(tǒng)一的一個Executors類中實現(xiàn)，由于能夠合理的使用CPU進(jìn)行對任務(wù)操作（并行操作），所以適合使用在很耗時的任務(wù)中
第2章 線程池原理分析
1、初始化
我們先看下初始化5個參數(shù)
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize, //主線程數(shù)
int maximumPoolSize, //最大線程數(shù)
long keepAliveTime, //線程存活時間 （除主線程外，其他的線程在沒有任務(wù)執(zhí)行的時候需要回收，多久后回收）
TimeUnit unit, //存活時間的時間單位
BlockingQueue<Runnable> workQueue, //阻塞隊列，我們需要執(zhí)行的task都在該隊列
ThreadFactory threadFactory, //生成thread的工廠
RejectedExecutionHandler handler) { //拒絕飽和策略，當(dāng)隊列滿了并且線程個數(shù)達(dá)到 maximunPoolSize 后采取的策略
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
2、execute方法
public void execute(Runnable command) {
if (command == null) //如果要執(zhí)行的任務(wù)是空的，異常
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();//111000...000
//高三位代表線程池的狀態(tài)，低29位代表線程池中的線程數(shù)量
//如果線程數(shù)小于主線程數(shù)，添加線程
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//如果超過主線程數(shù)，將任務(wù)添加至workqueue 阻塞隊列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//再判斷一次運行狀態(tài)，如果線程池不處于running狀態(tài)，則把剛加進(jìn)隊列的任務(wù)移除，如果移除成功則往下走進(jìn)行拒絕
if (! isRunning(recheck) && remove(command))
reject(command);
//接著上一個條件，如果移除失敗則判斷是否有工作線程，如果當(dāng)前線程池線程空，則添加一個線程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果超過主線程數(shù)且添加阻塞隊列失敗，則增加非核心線程，如果添加非核心線程也失敗，則拒絕
else if (!addWorker(command, false))
reject(command);
}
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//111000...000
private static final int COUNT_BITS = Integer.SIZE - 3;//29
private static final int CAPACITY = (1 << COUNT_BITS) - 1;//00011111 11111111 11111111 11111111
//00000000 00000000 00000000 00000001 << 29 =
//00100000 00000000 00000000 00000000 -1 =
//00011111 11111111 11111111 11111111
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS; //11100000 ... 000
//-1 原碼: 10000000 00000000 00000000 00000001
//-1 反碼: 11111111 11111111 11111111 11111110
//-1 補(bǔ)碼: 11111111 11111111 11111111 11111111 <<29=
// 11100000 0000000 00000000 00000000
private static final int SHUTDOWN = 0 << COUNT_BITS;//00000000 ... 000
private static final int STOP = 1 << COUNT_BITS;//001 0000 ... 000
private static final int TIDYING = 2 << COUNT_BITS;//010 0000 ... 000
private static final int TERMINATED = 3 << COUNT_BITS;//011 0000 ... 000
1、RUNNING
(1) 狀態(tài)說明：線程池處在RUNNING狀態(tài)時，能夠接收新任務(wù)，以及對已添加的任務(wù)進(jìn)行處理。
(02) 狀態(tài)切換：線程池的初始化狀態(tài)是RUNNING。換句話說，線程池被一旦被創(chuàng)建，就處于RUNNING狀態(tài)，并且線程池中的任務(wù)數(shù)為0！
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
2、 SHUTDOWN
(1) 狀態(tài)說明：線程池處在SHUTDOWN狀態(tài)時，不接收新任務(wù)，但能處理已添加的任務(wù)。
(2) 狀態(tài)切換：調(diào)用線程池的shutdown()接口時，線程池由RUNNING -> SHUTDOWN。
3、STOP
(1) 狀態(tài)說明：線程池處在STOP狀態(tài)時，不接收新任務(wù)，不處理已添加的任務(wù)，并且會中斷正在處理的任務(wù)。
(2) 狀態(tài)切換：調(diào)用線程池的shutdownNow()接口時，線程池由(RUNNING or SHUTDOWN ) -> STOP。
4、TIDYING
(1) 狀態(tài)說明：當(dāng)所有的任務(wù)已終止，ctl記錄的”任務(wù)數(shù)量”為0，線程池會變?yōu)門IDYING狀態(tài)。當(dāng)線程池變?yōu)門IDYING狀態(tài)時，會執(zhí)行鉤子函數(shù)terminated()。terminated()在ThreadPoolExecutor類中是空的，若用戶想在線程池變?yōu)門IDYING時，進(jìn)行相應(yīng)的處理；可以通過重載terminated()函數(shù)來實現(xiàn)。
(2) 狀態(tài)切換：當(dāng)線程池在SHUTDOWN狀態(tài)下，阻塞隊列為空并且線程池中執(zhí)行的任務(wù)也為空時，就會由 SHUTDOWN -> TIDYING。
當(dāng)線程池在STOP狀態(tài)下，線程池中執(zhí)行的任務(wù)為空時，就會由STOP -> TIDYING。
5、 TERMINATED
(1) 狀態(tài)說明：線程池徹底終止，就變成TERMINATED狀態(tài)。
(2) 狀態(tài)切換：線程池處在TIDYING狀態(tài)時，執(zhí)行完terminated()之后，就會由 TIDYING -> TERMINATED。
private static int runStateOf(int c){ return c & ~CAPACITY; }
private static int workerCountOf(int c){ return c & CAPACITY; }//CAPACITY:000111...111
private static int ctlOf(int rs, int wc){ return rs | wc; }
3、addWorker方法
private boolean addWorker(Runnable firstTask, boolean core){
retry: //goto語句 叫demo
//自旋檢查線程池的狀態(tài)。阻塞隊列是否為空等判斷
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))//如果線程池的運行狀態(tài)是running的話直接跳過該條件語句往下走，如果是>=SHUTDOWN的話就往后判斷（為什么不直接返回false不讓他創(chuàng)建worker呢，因為在shutdown狀態(tài)是可以創(chuàng)建線程去處理阻塞隊列里的任務(wù)的）
//此時因為rs>=SHTDOWN了，所以會先判斷是否等于SHUTDOWN，如果不等于就直接返回false不讓創(chuàng)建worker，如果等于的話接著往下判斷
//如果當(dāng)前任務(wù)不為空直接返回false不讓創(chuàng)建worker，（這里為什么當(dāng)前任務(wù)為空就直接不讓創(chuàng)建worker呢，就是因為shutdown狀態(tài)不能再接收新任務(wù)。
//如果當(dāng)前任務(wù)為空則判斷阻塞隊列是否為空，如果為空則返回false，不讓創(chuàng)建worker，如果不為空就不走這個條件，接著往下走
return false;
//自旋
for (;;) {
int wc = workerCountOf(c);
//如果現(xiàn)有線程數(shù)大于最大值，或者大于等于最大線程數(shù)（主線程數(shù)）
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//cas添加線程
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
//如果失敗了，繼續(xù)外層循環(huán)判斷
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//開啟一個線程，Worker實現(xiàn)了runnable接口
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//添加至wokers
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//添加成功
if (workerAdded) {
t.start(); //啟動線程，會調(diào)用我們線程的run接口，也就是我們worker的run
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
4、 goto語句demo
retry:
for (int i = 0; i < 3; i++) {
for (int j = 3; j < 10; j++) {
// if (j == 4) {
// break retry; //跳出外面循環(huán)
// }
if (j == 7) {
continue retry; //繼續(xù)外面循環(huán)
}
System.out.println(i+":"+j);
}
}
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker 禁止中斷，直到runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
5、worker.run方法
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try { //只要一直能獲取到task，就一直會執(zhí)行，不會關(guān)閉，所以線程也不會銷毀，線程銷毀只有當(dāng)task為null
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//調(diào)用線程方法之前執(zhí)行
beforeExecute(wt, task);
Throwable thrown = null;
try {
//調(diào)用task的run方法
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
//調(diào)用線程方法之后執(zhí)行
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
6、getTask()方法
private Runnable getTask(){
boolean timedOut = false; // Did the last poll() time out? //自旋獲取
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary. 必要時檢查空，狀態(tài)是否停止或者shutdown
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//獲取線程數(shù)量
int wc = workerCountOf(c);
// Are workers subject to culling?
//線程數(shù)大于主線程數(shù)時，或者allowCoreThreadTimeOut參數(shù)為true allowCoreThreadTimeOut默認(rèn)為false
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//超過最大線程，或者timed為true ,&& wc大于1個，并且任務(wù)隊列為空的時候
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
//線程數(shù)-1，并且返回null，該線程結(jié)束
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//如果time是true，超過時間不阻塞，不然一直阻塞，不回收
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//移除并返回隊列頭部的元素，如果為空，超過時間返回null
workQueue.take();
//移除并返回隊列頭部的元素，如果為空,一直阻塞
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
責(zé)任編輯：姜華
來源：
今日頭條
