ThreadPoolExecutor源码历险-主流程常用方法-e
2021-01-14 本文已影响0人
于情于你
ThreadPoolExecutor主流程常用方法execute
ThreadPoolExecutor核心方法，execute
下面的代码对应的线程池逻辑是，如果工作线程数小于核心线程数，启动一个新线程执行任务。否则将任务加入队列，如果队列满了那么启动非核心线程执行任务。如果队列满了，并且启动非核心线程失败，那么使用拒绝策略拒绝任务。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
// 如果工作线程数小于核心线程数，启动一个新线程执行任务
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// 如果任务入队成功，要进行二次检查。以防止在入队成功之后，线程池关闭了。
// 在必要的时候要回滚队列，拒绝任务。或者在没有线程的情况下启动新线程。
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果我们无法将任务排队，则尝试添加一个新线程。如果失败，我们知道我们已关闭或已已满了，因此拒绝该任务。
else if (!addWorker(command, false))
reject(command);
}
execute核心方法 addWorker
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 线程池已经准备要关闭了，不再接受新任务。可参考jdk源码中作者的注释：
// SHUTDOWN: Don't accept new tasks, but process queued tasks
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;
// CAS更新ctl的值+1，主要是给新线程预留空间
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 不管是更新失败还是还是线程池运行状态发生了变化，都要重试CAS更新
if (runStateOf(c) != rs)
continue retry;
}
}
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()) // 如果添加的任务已经开始跑了则抛出异常
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;
}
读完addWorker代码之后发现，其调用的主要方法有两个：一是添加线程成功后执行任务调用的Thread的start()方法。二是添加线程失败后调用的addWorkerFailed()处理方法。
下面来看一下这两个方法，调用线程的start方法之后还是会调用Worker的run方法。在上一篇的“不可变内部类Worker”的代码块里面可以看到run方法调用了runWorker方法。下面贴一下runWorker的代码和注释
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
// 任务是否正常执行完毕，如果正常执行完毕是false。非正常是true
// 因为一旦发生异常，异常会往上抛，阻断completedAbruptly = false的执行
boolean completedAbruptly = true;
try {
// 注意：如果执行的任务是空，则用getTask()方法从任务队列中拿一个任务
while (task != null || (task = getTask()) != null) {
w.lock();
// 确保线程池停止，线程被中断
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 任务执行前的hook方法，我们可以用子类扩展ThreadPoolExecutor
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 {
// 任务执行前的hook方法，我们可以用子类扩展ThreadPoolExecutor
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
可以看到runWorker方法里面有两个个钩子方法，分别是beforeExecute、afterExecute。这两个方法可以供我们通过子类扩展ThreadPoolExecutor，比如在afterExecute方法里面加一些任务的执行时间之类的日志。
另外，还需要重点说一下getTask方法。
getTask方法就是在工作队列workQueue里面拿任务执行，
如果此工作程序由于以下任何原因而必须退出，getTask则返回null：
1.当前的Worker数超过了maximumPoolSize
2.当前的线程池处于STOP状态
3.当前的线程池处于SHUTDOWN状态并且队列为空
4.这个Worker等待Task超时了，并且超时的Worker会被终止，也就是会判断allowCoreThreadTimeOut||workerCount>corePoolSize
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 当前的线程池处于STOP状态 或者当前的线程池处于SHUTDOWN状态并且队列为空
// 会减少Worker总数
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// timed 代表线程是否可以被回收
// allowCoreThreadTimeOut允许核心线程被回收
// wc > corePoolSize 线程数大于核心线程，回收非核心线程
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 当前的Worker数超过了maximumPoolSize或者Worker等待Task超时了
// CAS的递减workerCount，如果CAS失败，则continue继续
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 如果线程可以被回收的话，那么就调用poll，否则就调用take
// 因为poll在指定的超时时间内没有拿到任务的话，会直接返回null，不会等待。
// take如果队列中没有任务，则会一直等待
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
最后两个需要说的方法是上面runWorker中任务执行完毕，调用的processWorkerExit方法和addWorker中添加任务失败执行的addWorkerFailed方法。
processWorkerExit开始清理并且标记一个即将销毁的Worker。
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// completedAbruptly代表是否任务执行异常
// 如果是的话，减少工作线程
if (completedAbruptly)
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 已经执行的任务总数累加
completedTaskCount += w.completedTasks;
// 删除这个worker
workers.remove(w);
} finally {
mainLock.unlock();
}
// 尝试终止线程池
tryTerminate();
// ===========是否要添加新线程处理任务START===========
// 只有运行状态<STOP的时候才会处理，也就是RUNNING、SHUTDOWN
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
// 如果不是意外退出的话，那么就会判断最少要保留的Worker数
// 如果allowCoreThreadTimeOut被设置为true的话
// 那么说明核心线程在设置的KeepAliveTime之后，也会被销毁，所以少保留的Worker数为0
// 如果最少保留的Worker数为0的话，那么就会判断当前的任务队列是否为空
// 如果任务队列不为空的话而且线程池没有停止，那么说明至少还需要1个线程继续将任务完成。
// 判断当前的Worker是否大于min，
// 也就是说当前的Worker总数大于最少需要的Worker数的话
// 那么就直接返回，因为剩下的Worker会继续从WorkQueue中获取任务执行。
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 如果当前运行的Worker数比当前所需要的Worker数少的话，
// 那么就会调用addWorker，添加新的Worker，也就是新开启线程继续处理任务。
addWorker(null, false);
// ===========是否要添加新线程处理任务END===========
}
}
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// =======判断是否有资格终止START=======
// 如果线程池还处于运行状态或者
// 已经在TIDYING或者之后了、在SHUTDOWN状态但是任务队列不为空，还有要执行的任务则不终止则不关闭
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 如果还有工作线程，说明当前线程不是最后一个执行线程，则不执行销毁操作
// 只要中断一个空闲线程就可以了，就是中断自己然后就返回
// 销毁操作交给最后一个执行任务的线程
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//CAS将ctl状态改为TIDYING
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// 钩子方法， 可以用子类扩展。当前ThreadPoolExecutor没有给他实现
terminated();
} finally {
//CAS将ctl状态改为TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
// 可能在此之前某线程调用了awaitTermination方法，一直处在阻塞中，
// 并且没有超时，也没有发生中断。那么在销毁线程池的此时就需要唤醒这些线程了
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
addWorkerFailed是addWorker添加新任务失败的处理方法，也可以看作是添加任务的回滚方法
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 在workers集合里面删除Worker
if (w != null)
workers.remove(w);
// 减少worker数量
decrementWorkerCount();
// 尝试关闭线程池
tryTerminate();
} finally {
mainLock.unlock();
}
}
execute流程所有的代码都在这了，对jdk作者为什么这样写，也在注释里面加了描述，可以看到调用tryTerminate的地方非常多。不管是线程执行完毕，还是添加线程失败都要尝试关闭一下。下一篇我们会说用的比较多的，submit方法.
