ThreadPoolExecutor类:
ThreadPoolExecutor是我们最常用的一个线程池类,它实现了AbstractExecutorService接口。首先来看一下它的构造器及相关关键变量:
// 这是其中的一个构造器,包含了线程池构造器的七大核心参数
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
this.corePoolSize = corePoolSize; // 核心线程数
this.maximumPoolSize = maximumPoolSize; // 最大线程数
this.workQueue = workQueue; // 任务队列,可选择链表LinkedBlockingQueue或数组ArrayBlockingQueue实现
this.keepAliveTime = unit.toNanos(keepAliveTime); // 非核心线程的存活时间,若要指定核心线程的存活时间,需设置allowCoreThreadTimeOut,// 存活时间单位
this.threadFactory = threadFactory; // 线程工厂,一般使用Executors工具类中的defaultThreadFactory
this.handler = handler; // 拒绝策略,包括四种,1. AbortPolicy抛出异常;2. CallerRunsPolicy由主线程自己执行任务;3. DiscardPolicy 直接丢弃任务;4. DiscardOldestPolicy 丢弃队列中等待时间最长的任务
}
// 关键变量:
// ctl是ThreadPoolExecutor类中的一个状态标记字段,使用int类型AtomicInteger类型值存放,其中高三位存放运行状态,低29位存放线程数量。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 运行状态存放在int类型高3位,共有五种状态:RUNNING,SHUDOWN,STOP,TIDYING,TERMINATED;注意其中只有RUNNING状态小于0;后面可据此判断是否为RUNNING状态。
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;
下面这张图介绍了ThreadPoolExecutor线程池的任务执行流程,即execute方法的执行流程
通过对比上图的执行流程,我们了解一下execute方法
public void execute(Runnable command) {
int c = ctl.get();
// 如果当前工作线程小于核心线程,则新增工作线程并返回
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) { // 如果当前执行器是running状态,并且将任务放入队列中
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command)) // 再次判断,如果执行器状态不是running,则将队列中的任务移除,并执行拒绝策略
reject(command);
else if (workerCountOf(recheck) == 0) // 如果工作线程数为0,则添加非核心线程
addWorker(null, false);
} else if (!addWorker(command, false)) //如果添加非核心线程失败,则执行拒绝策略
reject(command);
}
相信通过上图结合方法中的注释内容,我们大体上了解了任务的执行流程,接着我们进一步了解一下addWorker方法:
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
// 如果执行器状态至少为SHUTDOWN并且运行状态至少为STOP或者任务不为空或者任务队列为空,则直接返回false
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
for (;;) {
// 如果当前线程数已经大于等于核心/最大线程数,则直接返回false
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
if (compareAndIncrementWorkerCount(c)) // 如果CAS工作线程加1成功,则跳出循环
break retry;
c = ctl.get(); // 再次获取ctl值
if (runStateAtLeast(c, SHUTDOWN)) // 如果执行器状态至少为SHUTDOWN,则继续执行循环
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 {
int c = ctl.get();
// 若为运行状态,workers数加一
if (isRunning(c) ||
(runStateLessThan(c, STOP) && 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(); // workers数添加成功了,则启动该线程执行任务
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w); // 线程启动失败,则移除workers中的该线程,同时ctl减一
}
return workerStarted;
}
由上面的addWork方法我们可以知道,若添加worker则执行run方法,而查看源码我们可以知道,在run方法中调用的是runWork方法,下面我们继续来看一下runWork方法:
// 任务执行主循环体
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) { // 存在未执行的任务
w.lock();
// 条件一. 线程至少处于STOP状态或者线程被中断并且至少处于STOP状态
// 条件二. 该线程没有被中断标记
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt(); // 中断该线程
try {
beforeExecute(wt, task); // 钩子方法,可自定义子类实现,清除ThreadLocals或者执行日志记录
try {
task.run();
afterExecute(task, null); // 钩子方法,自定义子类实现
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
在上面的runWork中采用了while的死循环方式,只要有未完成的任务就去执行,在这里是通过getTask方法来不断的获取队列中的任务,那接下来我们看一下getTask方法:
// 阻塞或定时等待获取任务
private Runnable getTask() {
boolean timedOut = false;
for (;;) {
int c = ctl.get();
// 如果执行器至少为SHUTDOWN状态,并且执行器至少到了STOP状态或者任务队列为空时,worker数减一,返回null值
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
// 如果worker数大于最大线程数或者超出存活时间,则线程数减一,返回null
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ? // 若线程数大于核心线程数,则有存活时间的等待获取队列中的任务,否则阻塞take任务
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
经过上面的流程,大体上可以了解了ThreadPoolExecutor线程池的执行过程。