了解Java线程池执行原理

前言

上一篇已经对进行了分析，了解线程池既有预设的模板，也提供多种参数支撑灵活的定制。

本文将会围绕线程池的生命周期，分析线程池执行任务的过程。

线程池状态

首先认识两个贯穿线程池代码的参数：

• runstate：线程池运行状态
• workercount：工作线程的数量

线程池用一个32位的int来同时保存runstate和workercount，其中高3位是runstate，其余29位是workercount。代码中会反复使用runstateof和workercountof来获取runstate和workercount。

private final atomicinteger ctl = new atomicinteger(ctlof(running, 0));
private static final int count_bits = integer.size - 3;
private static final int capacity = (1 << count_bits) - 1;
// 线程池状态
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;
// ctl操作
private static int runstateof(int c) { return c & ~capacity; }
private static int workercountof(int c) { return c & capacity; }
private static int ctlof(int rs, int wc) { return rs | wc; }

• running：可接收新任务，可执行等待队列里的任务
• shutdown：不可接收新任务，可执行等待队列里的任务
• stop：不可接收新任务，不可执行等待队列里的任务，并且尝试终止所有在运行任务
• tidying：所有任务已经终止，执行terminated()
• terminated：terminated()执行完成

线程池状态默认从running开始流转，到状态terminated结束，中间不需要经过每一种状态，但不能让状态回退。下面是状态变化可能的路径和变化条件：

worker的创建

线程池是由worker类负责执行任务，worker继承了abstractqueuedsynchronizer，引出了java并发框架的核心aqs。

abstractqueuedsynchronizer，简称aqs，是java并发包里一系列同步工具的基础实现，原理是根据状态位来控制线程的入队阻塞、出队唤醒来处理同步。

aqs不会在这里展开讨论，只需要知道worker包装了thread，由它去执行任务。

调用execute将会根据线程池的情况创建worker，可以归纳出下图四种情况：

public void execute(runnable command) {
if (command == null)
throw new nullpointerexception();
int c = ctl.get();
//1
if (workercountof(c) < corepoolsize) {
if (addworker(command, true))
return;
c = ctl.get();
}
//2
if (isrunning(c) && workqueue.offer(command)) {
int recheck = ctl.get();
if (! isrunning(recheck) && remove(command))
//3
reject(command);
else if (workercountof(recheck) == 0)
//4
addworker(null, false);
}
//5
else if (!addworker(command, false))
//6
reject(command);
}

标记1对应第一种情况，要留意addworker传入了core，core=true为corepoolsize，core=false为maximumpoolsize，

新增时需要检查workercount是否超过允许的最大值。

标记2对应第二种情况，检查线程池是否在运行，并且将任务加入等待队列。标记3再检查一次线程池状态，如果线程池忽然处于非运行状态，那就将等待队列刚加的任务删掉，再交给rejectedexecutionhandler处理。标记4发现没有worker，就先补充一个空任务的worker。

标记5对应第三种情况，等待队列不能再添加任务了，调用addworker添加一个去处理。

标记6对应第四种情况，addworker的core传入false，返回调用失败，代表workercount已经超出maximumpoolsize，那就交给rejectedexecutionhandler处理。

private boolean addworker(runnable firsttask, boolean core) {
//1
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
}
}
//2
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;
}

标记1的第一段代码，目的很简单，是为workercount加一。至于为什么代码写了这么长，是因为线程池的状态在不断

变化，并发环境下需要保证变量的同步性。外循环判断线程池状态、任务非空和队列非空，内循环使用cas机制保证workercount正确地递增。不了解cas可以看认识非阻塞的同步机制cas，后续增减workercount都会使用cas。

标记2的第二段代码，就比较简单。创建一个新worker对象，将worker添加进workers里（set集合）。成功添加后，启动worker里的线程。在finally里判断线程是否启动成功，不成功直接调用addworkerfailed。

private void addworkerfailed(worker w) {
final reentrantlock mainlock = this.mainlock;
mainlock.lock();
try {
if (w != null)
workers.remove(w);
decrementworkercount();
tryterminate();
} finally {
mainlock.unlock();
}
}

addworkerfailed将减少已经递增的workercount，并且调用tryterminate结束线程池。

worker的执行

worker(runnable firsttask) {
setstate(-1); // inhibit interrupts until runworker
this.firsttask = firsttask;
this.thread = getthreadfactory().newthread(this);
}
public void run() {
runworker(this);
}

worker在构造函数里采用threadfactory创建thread，在run方法里调用了runworker，看来是真正执行任务的地方。

final void runworker(worker w) {
thread wt = thread.currentthread();
runnable task = w.firsttask;
w.firsttask = null;
w.unlock(); // allow interrupts
boolean completedabruptly = true;
try {
//1
while (task != null || (task = gettask()) != null) {
w.lock();
//2
if ((runstateatleast(ctl.get(), stop) ||
(thread.interrupted() &&
runstateatleast(ctl.get(), stop))) &&
!wt.isinterrupted())
wt.interrupt();
try {
//3
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;
//4
w.completedtasks++;
w.unlock();
}
}
completedabruptly = false; //5
} finally {
//6
processworkerexit(w, completedabruptly);
}
}

标记1进入循环，从gettask获取要执行的任务，直到返回null。这里达到了线程复用的效果，让线程处理多个任务。

标记2是一个比较复杂的判断，保证了线程池在stop状态下线程是中断的，非stop状态下线程没有被中断。如果你不了解java的中断机制，看如何正确结束java线程这篇。

标记3调用了run方法，真正执行了任务。执行前后提供了beforeexecute和afterexecute两个方法，由子类实现。

标记4里的completedtasks统计worker执行了多少任务，最后累加进completedtaskcount变量，可以调用相应方法返回一些统计信息。

标记5的变量completedabruptly表示worker是否异常终止，执行到这里代表执行正常，后续的方法需要这个变量。

标记6调用processworkerexit结束，后面会分析。

接着来看worker从等待队列获取任务的gettask方法：

private runnable gettask() {
boolean timedout = false; // did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runstateof(c);
//1
// check if queue empty only if necessary.
if (rs >= shutdown && (rs >= stop || workqueue.isempty())) {
decrementworkercount();
return null;
}
int wc = workercountof(c);
//2
// 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;
}
//3
try {
runnable r = timed ?
workqueue.poll(keepalivetime, timeunit.nanoseconds) :
workqueue.take();
if (r != null)
return r;
timedout = true;
} catch (interruptedexception retry) {
timedout = false;
}
}
}

标记1检查线程池的状态，这里就体现出shutdown和stop的区别。如果线程池是shutdown状态，还会先处理完等待队列的任务；如果是stop状态，就不再处理等待队列里的任务了。

标记2先看allowcorethreadtimeout这个变量，false时worker空闲，也不会结束；true时，如果worker空闲超过keepalivetime，就会结束。接着是一个很复杂的判断，好难转成文字描述，自己看吧。注意一下wc>maximumpoolsize，出现这种可能是在运行中调用setmaximumpoolsize，还有wc>1，在等待队列非空时，至少保留一个worker。

标记3是从等待队列取任务的逻辑，根据timed分为等待keepalivetime或者阻塞直到有任务。

最后来看结束worker需要执行的操作：

private void processworkerexit(worker w, boolean completedabruptly) {
//1
if (completedabruptly) // if abrupt, then workercount wasn't adjusted
decrementworkercount();
//2
final reentrantlock mainlock = this.mainlock;
mainlock.lock();
try {
completedtaskcount += w.completedtasks;
workers.remove(w);
} finally {
mainlock.unlock();
}
//3
tryterminate();
int c = ctl.get();
//4
if (runstatelessthan(c, stop)) {
if (!completedabruptly) {
int min = allowcorethreadtimeout ? 0 : corepoolsize;
if (min == 0 && ! workqueue.isempty())
min = 1;
if (workercountof(c) >= min)
return; // replacement not needed
}
addworker(null, false);
}
}

正常情况下，在gettask里就会将workercount减一。标记1处用变量completedabruptly判断worker是否异常退出，如果是，需要补充对workercount的减一。

标记2将worker处理任务的数量累加到总数，并且在集合workers中去除。

标记3尝试终止线程池，后续会研究。

标记4处理线程池还是running或shutdown状态时，如果worker是异常结束，那么会直接addworker。如果allowcorethreadtimeout=true，并且等待队列有任务，至少保留一个worker；如果allowcorethreadtimeout=false，workercount不少于corepoolsize。

总结一下worker：线程池启动后，worker在池内创建，包装了提交的runnable任务并执行，执行完就等待下一个任务，不再需要时就结束。

线程池的关闭

线程池的关闭不是一关了事，worker在池里处于不同状态，必须安排好worker的”后事”，才能真正释放线程池。threadpoolexecutor提供两种方法关闭线程池：

• shutdown：不能再提交任务，已经提交的任务可继续运行；
• shutdownnow：不能再提交任务，已经提交但未执行的任务不能运行，在运行的任务可继续运行，但会被中断，返回已经提交但未执行的任务。

public void shutdown() {
final reentrantlock mainlock = this.mainlock;
mainlock.lock();
try {
checkshutdownaccess(); //1 安全策略机制
advancerunstate(shutdown); //2
interruptidleworkers(); //3
onshutdown(); //4 空方法，子类实现
} finally {
mainlock.unlock();
}
tryterminate(); //5
}

shutdown将线程池切换到shutdown状态，并调用interruptidleworkers请求中断所有空闲的worker，最后调用tryterminate尝试结束线程池。

public list<runnable> shutdownnow() {
list<runnable> tasks;
final reentrantlock mainlock = this.mainlock;
mainlock.lock();
try {
checkshutdownaccess();
advancerunstate(stop);
interruptworkers();
tasks = drainqueue(); //1
} finally {
mainlock.unlock();
}
tryterminate();
return tasks;
}

shutdownnow和shutdown类似，将线程池切换为stop状态，中断目标是所有worker。drainqueue会将等待队列里未执行的任务返回。

interruptidleworkers和interruptworkers实现原理都是遍历workers集合，中断条件符合的worker。

上面的代码多次出现调用tryterminate，这是一个尝试将线程池切换到terminated状态的方法。

final void tryterminate() {
for (;;) {
int c = ctl.get();
//1
if (isrunning(c) ||
runstateatleast(c, tidying) ||
(runstateof(c) == shutdown && ! workqueue.isempty()))
return;
//2
if (workercountof(c) != 0) { // eligible to terminate
interruptidleworkers(only_one);
return;
}
//3
final reentrantlock mainlock = this.mainlock;
mainlock.lock();
try {
if (ctl.compareandset(c, ctlof(tidying, 0))) {
try {
terminated();
} finally {
ctl.set(ctlof(terminated, 0));
termination.signalall();
}
return;
}
} finally {
mainlock.unlock();
}
// else retry on failed cas
}
}

标记1检查线程池状态，下面几种情况，后续操作都没有必要，直接return。

• running（还在运行，不能停）
• tidying或terminated（已经没有在运行的worker）
• shutdown并且等待队列非空（执行完才能停）

标记2在worker非空的情况下又调用了interruptidleworkers，你可能疑惑在shutdown时已经调用过了，为什么又调用，而且每次只中断一个空闲worker？

你需要知道，shutdown时worker可能在执行中，执行完阻塞在队列的take，不知道要结束，所有要补充调用interruptidleworkers。每次只中断一个是因为processworkerexit时，还会执行tryterminate，自动中断下一个空闲的worker。

标记3是最终的状态切换。线程池会先进入tidying状态，再进入terminated状态，中间提供了terminated这个空方法供子类实现。

调用关闭线程池方法后，需要等待线程池切换到terminated状态。awaittermination检查限定时间内线程池是否进入terminated状态，代码如下：

public boolean awaittermination(long timeout, timeunit unit)
throws interruptedexception {
long nanos = unit.tonanos(timeout);
final reentrantlock mainlock = this.mainlock;
mainlock.lock();
try {
for (;;) {
if (runstateatleast(ctl.get(), terminated))
return true;
if (nanos <= 0)
return false;
nanos = termination.awaitnanos(nanos);
}
} finally {
mainlock.unlock();
}
}

后言

以上过了一遍线程池主要的逻辑，总体来看线程池的设计是很清晰的。如有错误或不足，欢迎指出，也欢迎留言交流。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。