addWorker(Runnable firstTask, boolean core)方法
/**
* Checks if a new worker can be added with respect to current
* pool state and the given bound (either core or maximum). If so,
* the worker count is adjusted accordingly, and, if possible, a
* new worker is created and started, running firstTask as its
* first task. This method returns false if the pool is stopped or
* eligible to shut down. It also returns false if the thread
* factory fails to create a thread when asked. If the thread
* creation fails, either due to the thread factory returning
* null, or due to an exception (typically OutOfMemoryError in
* Thread.start()), we roll back cleanly.
*
* @param firstTask the task the new thread should run first (or
* null if none). Workers are created with an initial first task
* (in method execute()) to bypass queuing when there are fewer
* than corePoolSize threads (in which case we always start one),
* or when the queue is full (in which case we must bypass queue).
* Initially idle threads are usually created via
* prestartCoreThread or to replace other dying workers.
*
* @param core if true use corePoolSize as bound, else
* maximumPoolSize. (A boolean indicator is used here rather than a
* value to ensure reads of fresh values after checking other pool
* state).
* @return true if successful
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
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()))
return false;
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;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
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();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
執(zhí)行流程
①首先兩個for循環(huán)，先來看外面的for循環(huán)
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null &&
! workQueue.isEmpty()))
return false;
1、首先獲取當(dāng)前線程池的運行狀態(tài)。進入2
2、先來回顧一個線程池的幾種狀態(tài)：
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;
if語句里主要判斷如果當(dāng)前線程池已經(jīng)被關(guān)閉了或者線程池被關(guān)閉了，緩存隊列還有任務(wù)，那直接返回false，結(jié)束流程。這里只要是判斷線程池是否已經(jīng)被關(guān)閉了，如果已經(jīng)被關(guān)閉了，那么就不再接受任務(wù)了。
③如果線程池還沒被關(guān)閉，那么進入第二for循環(huán)
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;
// else CAS failed due to workerCount change; retry inner loop
}
1、首先獲取到當(dāng)前線程池的線程數(shù)
int wc = workerCountOf(c);
2、判斷線程數(shù)有沒有>= corePoolSize : maximumPoolSize，成立的話，直接退出循環(huán)，返回false。下面的core是boolean值
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
return false;
3、然后更新線程池的線程數(shù)
if (compareAndIncrementWorkerCount(c))
break retry;
更新不成功，則結(jié)束本循環(huán)
4、再次判斷線程池的運行狀態(tài)
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
如果不相等，說明線程池的狀態(tài)已經(jīng)被其他線程改變了，直接結(jié)束此次循環(huán)
上面的兩個循環(huán)只要是判斷線程池的狀態(tài)，然后更新線程池的線程數(shù)
接下來的流程就是把任務(wù)添加到任務(wù)隊列中，并交給線程池中空閑的線程來執(zhí)行
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
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();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
5、首先創(chuàng)建一個worker工作隊列對象
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
可以看出，這個對象保存提交的任務(wù)，并把創(chuàng)建一個工作線程來執(zhí)行這個提交的任務(wù)
6、首先加鎖，然后判斷線程池狀態(tài)，如果處于運行時狀態(tài)，就把剛才創(chuàng)建的worker工作隊列添加到worker工作隊列中，然后啟動工作線程執(zhí)行任務(wù)
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();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
Worker
線程池中的每一個線程被封裝成一個Worker對象，ThreadPool維護的其實就是一組Worker對象，看一下Worker的定義：
private final class Worker extends AbstractQueuedSynchronizer implements Runnable
可以看到worker本身就是一個線程，所以看它的run（）方法
public void run() {
runWorker(this);
}
最終調(diào)用了 runWorker(this)方法
/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
*
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is
* running, we get tasks from getTask. If it returns null then the
* worker exits due to changed pool state or configuration
* parameters. Other exits result from exception throws in
* external code, in which case completedAbruptly holds, which
* usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent
* other pool interrupts while the task is executing, and then we
* ensure that unless pool is stopping, this thread does not have
* its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which
* might throw an exception, in which case we cause thread to die
* (breaking loop with completedAbruptly true) without processing
* the task.
*
* 4. Assuming beforeExecute completes normally, we run the task,
* gathering any of its thrown exceptions to send to afterExecute.
* We separately handle RuntimeException, Error (both of which the
* specs guarantee that we trap) and arbitrary Throwables.
* Because we cannot rethrow Throwables within Runnable.run, we
* wrap them within Errors on the way out (to the thread's
* UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate
* information as we can provide about any problems encountered by
* user code.
*
* @param w the worker
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 獲取第一個任務(wù)
Runnable task = w.firstTask;
w.firstTask = null;
// 允許中斷
w.unlock(); // allow interrupts
// 是否因為異常退出循環(huán)
boolean completedAbruptly = true;
try {
// 如果task為空，則通過getTask來獲取任務(wù)
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 {
beforeExecute(wt, task);
Throwable thrown = null;
try {
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 {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
執(zhí)行流程
①while循環(huán)不斷地通過getTask()方法獲取任務(wù)；
②getTask()方法從阻塞隊列中取任務(wù)；
③如果線程池正在停止，那么要保證當(dāng)前線程是中斷狀態(tài)，否則要保證當(dāng)前線程不是中斷狀態(tài)；
④調(diào)用task.run()執(zhí)行任務(wù)；
⑤如果task為null則跳出循環(huán)，執(zhí)行processWorkerExit()方法；
⑥r(nóng)unWorker方法執(zhí)行完畢，也代表著Worker中的run方法執(zhí)行完畢，銷毀線程。
