上一篇分析了ThreadPoolExecutor的execute方法的具体执行过程，这一篇主要分析当中的几个重要的函数。
http://www.jianshu.com/p/f992654ef203
- private boolean addWorker(Runnable firstTask, boolean core)
首先分析一下参数，firstTask就是指我们用户传入的需要执行的认为，core参数，当为真时，表示当前线程数小于corePoolSize，为假就是线程数大于corePoolSize。该方法主要有两块大的逻辑，第一部分是通过CAS操作去跟新当前的线程数量：
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
}
}
两个大的for无限循环体，外边的循环就是不断检测当前线程池的状态，一旦发现不处于Running状态了，就立刻返回false了，因为通过前边文章分析我们知道，当线程池调用了shutdown方法后，就不再接收新的任务了。
再看内层的循环体，主要就是通过CAS操作去更新线程池数量，
if (compareAndIncrementWorkerCount(c))如果更新成功，那么就会跳出外层循环，继续后边的逻辑。如果更新失败，需要再次检查当前线程池的状态，因为CAS操作是没有进行同步操作的，所以有可能此时线程池的状态已经变了，需要再次检测。发现状态发生变化，那么需要跳出内层循环直接进行下一次的外层循环，重新检查线程池状态。如果线程池的状态没有发生变化，则继续下一次的内层循环，重新尝试更新线程池数量直至更新成功。
线程池数量更新成功了，就进入下一个逻辑块，添加线程：
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;
}
第5行首先新建了一个worker，这个类内部封装了我们的runnable对象还有线程对象（随后会进行分析），第9行进行加锁，在线程池操作中，所有涉及到线程池状态变化的代码都需要进行同步操作，前一部分用的是CAS，这一部分用到了可重入锁。
在判断线程池是处于running状态后，在20行workers.add(w)，把工作线程添加到一个Set集合中，workers就是一个HashSet用于存放当前线程池中存活的线程。后边就很简单，记录了一下最大的线程数。
关键看30行，t.start()启动了线程。这个t就是我们新建的worker中的线程。这样一下来我们在线程数小于corePoolSize的时候，发起的任务就直接通过新建线程开始运行了。
为了了解发出的任务到底是怎么运行的，还得继续分析Worker这个类。
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
...//省略了重写AbstractQueuedSynchronizer类中的方法
}
这个worker还是继承了Runnable接口，内部有三个成员变量，注释中已经写得很明白都要什么作用。我们关注一下run方法，其实就调用了一个runWorker(this)，把自己作为参数传递进去。那么继续跟进这个方法（请保持清醒，因为已经有好几层调用了。。）：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
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);
}
}
主体是一个while循环，判断当前task是否为空，不为空就直接运行此任务，关键代码在第19行task.run()，直接调用Runnable的run方法，那就是在当前线程运行它，那么问题来了，现在当前线程到底是哪个线程了？是不是有点乱了，赶紧来个流程图，我们观察一下函数调用：
方法调用过程
第一二步都是在用户线程（调用execute方法时的那个线程），后边三步之后已经切换到新建的worker线程中去了，所以最终用户传递给execute的那个Runnable对象是在新的线程中运行的。
ok！回到上边代码的while循环，或条件判断的第二个条件回去调用getTask方法。看名字都知道，肯定是去BlockingQueue（任务缓存队列）中去取下一个需要执行的任务了。
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
第7行的boolean timed = allowCoreThreadTimeOut || wc > corePoolSize，主要是判断当前线程是否需要进行销毁，allowCoreThreadTimeOut 为真，就是说不管当前的线程数有没有超过corePoolSize，都需要根据keepAliveTime去判断线程的存活期。如果为假那么就看第二个语句，就是当前线程数大于corePoolSize就需要判断线程的存活期，如果不大于，那么就不需要判断存活期了。
接下来主要关注一下16-25行的try代码块，需要判断线程的存活期那么r就为workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)返回的值，也就是说尝试从缓存队列取任务，时间限制为keepAliveTime，如果在这段时间内取到了任务，就返回，如果没取到那么会抛出异常，那么就表明这个线程已经在keepAliveTime这个时间段里没有执行了，那么该线程就该被销毁了。
time为false的话，就表明不需要判断线程的存活期，直接调用workQueue.take()，由于线程池采用的阻塞队列，那么该线程就会一直阻塞，直到从队列中取到新的任务，然后返回。
到此我们的ThreadPoolExecutor的大致流程就分析完毕了，其中还有很多细节就没法完全覆盖到了，如果有理解不当的地方，欢迎大家指出，共同学习交流！
网友评论