一、线程池状态
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;
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS; // 111
private static final int SHUTDOWN = 0 << COUNT_BITS; // 000
private static final int STOP = 1 << COUNT_BITS; // 001
private static final int TIDYING = 2 << COUNT_BITS; // 010
private static final int TERMINATED = 3 << COUNT_BITS; // 011
// Packing and unpacking 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：整理。所有任务已终止workCount为0，将执行terminated()钩子方法。在ThreadPoolExecutor中可以对terminated()进行重载。
- TERMINATED：彻底结束。terminated()方法执行完成。
graph LR;
RUNNING(RUNNING) --shutdown--> SHUTDOWN(SHUTDOWN)
RUNNING --shutdownNow--> STOP(STOP)
SHUTDOWN --> |阻塞队列为空,执行任务为空| TIDYING(TIDYING)
STOP -->|线程池中执行任务为空| TIDYING
TIDYING --terminated--> TERMINATED(TERMINATED)
ctl 值解析
二进制前3位表示线程池状态，后29位表示工作线程数量。
二、创建线程池
线程池创建主要通过构造函数，有7个参数。
- corePoolSize：核心线程数。
- maximumPoolSize：最大线程数。
- keepAliveTime：线程存活时间。
- unit：时间单位。
- workQueue：阻塞队列。
- threadFactory：线程工厂。
- handler：拒绝策略，四种。
- AbortPolicy：直接抛出异常，默认策略；
- CallerRunsPolicy：回退给调用者所在的线程来执行任务；
- DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
- DiscardPolicy：直接丢弃任务；
- 当然我们也可以实现自己的拒绝策略，实现RejectedExecutionHandler接口即可。
三、终止线程池
线程池ThreadPoolExecutor提供了shutdown()和shutDownNow()两种方式关闭线程池。
- shutdown()：按过去执行已提交任务的顺序发起一个有序的关闭，但是不接受新任务。
- shutdownNow() :尝试停止所有的活动执行任务、暂停等待任务的处理，并返回等待执行的任务列表。
四、核心源码阅读
1. 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)) {
// double check 线程池状态
int recheck = ctl.get();
// 线程池不是running则回滚队列
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
// 没有工作线程，则创建线程
addWorker(null, false);
}
// 尝试将当任务程加入非核心线程
else if (!addWorker(command, false))
// 拒绝当前任务
reject(command);
}
总结如下：
首先判断工作线程少于核心线程，则尝试将当前任务添加为核心线程，添加成功就返回；
-
未能成功添加核心线程，看线程池状态如果是RUNNING，则尝试将任务加入等待队列；
如果加入成功，则需要对当前任务负责，进行double check是为了保证已经加入队列的任务要么等待线程来处理它，要么回退拒绝。
- 再次检查线程池状态，如果发现线程池不是
running了，则回退任务执行拒绝策略； - 再次检查线程池内工作线程数量，没有工作线程则创建工作线程，用于执行队列中的剩余任务；
- 再次检查线程池状态，如果发现线程池不是
未能成功加入等待队列，则尝试将当前任务加入非核心线程，失败则执行拒绝策略；
2. addWorker()添加工作线程
private boolean addWorker(Runnable firstTask, boolean core) {
// 1. 尝试通过CAS增加工作线程数量
retry:
for (;;) {
int c = ctl.get();
// 线程池状态
int rs = runStateOf(c);
// 非RUNNING：SHUTDOWN,STOP,TIDYING,TERMINATED
// SHUTDOWN状态不允许添加新任务，只能处理队列中已有任务。
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增加线程数量
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
// 线程池状态改变，则去外层循环重新检测状态
if (runStateOf(c) != rs)
continue retry;
}
}
// 2. 创建工作线程，创建失败则回滚。
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 新建worker线程，通过firstTask
w = new Worker(firstTask);
// worker中通过工厂创建的线程，当前线程
final Thread t = w.thread;
// 线程创建成功
if (t != null) {
// 获取主锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 线程池状态
int rs = runStateOf(ctl.get());
// running || (shutdown && firstTask == null)
// running允许添加任务，shutdown不允许添加任务（队列中可能有未执行任务，创建没有firstTask工作线程处理）
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
// 当前线程已经启动，抛出异常
if (t.isAlive())
throw new IllegalThreadStateException();
// private final HashSet<Worker> workers = new HashSet<Worker>();
workers.add(w);
// 修改最大线程数量和线程添加成功变量
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 线程添加成功，则启动线程
if (workerAdded) {
// 启动，执行worker的run方法
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
// 未能成功启动，则走失败回滚
addWorkerFailed(w);
}
return workerStarted;
}
添加工作线程比较简单，首先是通过CAS对线程数量进行增加。然后尝试创建线程，如未能成功创建则走回滚方法。
首先要判断线程线程池状态，如果是非RUNNING状态，则不能进行添加线程。这里要注意SHUTDOWN状态的时候，不能添加新任务，但是可以对阻塞队列中已有的任务进行处理。是通过这段代码判断的(rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))。
所以，只有两种情况能创建工作线程，即：线程池RUNNING状态，线程池SHUTDOWN状态并且没有firstTask的时候。
后者创建出的工作线程，用于处理阻塞队列中的未执行的任务，但是此时已经不接收新任务了。与excutor方法中的double check处代码对应。
在executor()方法中新建核心线程的地方有3处：
- 核心线程数量不够时，创建核心线程；
- 把当前任务加入队列后，未能从队列中回滚。创建一个没有firstTask的工作线程；
- 不是RUNNING或未能加入到阻塞队列，则创建非核心线程；
3. addWorkerFailed()添加失败回滚
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// removes worker from workers, if present
if (w != null)
workers.remove(w);
// decrements worker count
decrementWorkerCount();
// rechecks for termination, in case the existence of this worker was holding up termination
tryTerminate();
} finally {
mainLock.unlock();
}
}
如果添加失败则需要回滚：
- 如果当前work存在，则删除。
- 把在尝试创建时添加的工作线程数量减下去。
- 尝试终止当前工作线程。
4. Worker 工作线程
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
private static final long serialVersionUID = 6138294804551838833L;
// 工厂创建出的线程，失败时候为null
final Thread thread;
// 将要运行的第一个任务，可能为null
Runnable firstTask;
// 完成任务计数器
volatile long completedTasks;
Worker(Runnable firstTask) {
// 设置AQS的同步状态private volatile int state，是一个计数器，大于0代表锁已经被获取
setState(-1);
this.firstTask = firstTask;
// 利用工厂创建线程
this.thread = getThreadFactory().newThread(this);
}
// 任务执行
public void run() {
runWorker(this);
}
// Lock methods 省略……
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
}
继承了AQS实现了Runnable，即能释放锁获取锁，又是一个可执行的任务。线程启动后执行run()方法，其实就是调用的外部runWorker方法。
5. runWorker() 任务执行方法
final void runWorker(Worker w) {
// 当前线程，即当前worker
Thread wt = Thread.currentThread();
// 要执行的第一个任务，可能为null
Runnable task = w.firstTask;
w.firstTask = null;
// 主要是为了把添加工作线程时的setState(-1)改为0，释放锁。
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 一直在获取要执行的任务，getTask从队列中获取。
while (task != null || (task = getTask()) != null) {
// worker 获取锁，要进行任务处理了
w.lock();
// 确保线程池STOP时线程中断，非STOP则清除中断。
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 {
// worker退出流程
processWorkerExit(w, completedAbruptly);
}
}
- 首先根据worker获取要执行的task，把添加工作线程时的setState(-1)改为0，释放锁。
- 然后循环从阻塞队列中获取任务，并执行任务的run方法。
- 在执行任务的前后分别有两个自定义方法，可供拓展。
6. getTask()队列中获取任务
private Runnable getTask() {
// 超时，用于关闭非核心线程
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 检测线程池状态，非RUNNING状态。SHUTDOWN并且队列中没有任务
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
// 减掉工作线程
decrementWorkerCount();
return 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 ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
主要就是在阻塞队列中取待处理的任务，当线程池中工作线程多于核心线程时，利用队列的poll获取，未达到核心数则直接take。
take方法没有任务时，则会一直阻塞挂起等待，直到有任务加入时唤起该线程，返回任务。
poll方法则是在规定时间范围内获取，如果未获取到则返回null。
7. processWorkerExit()工作线程退出
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// true：用户线程运行异常,需要扣减线程数
// false：正常退出的，在getTask方法中扣减的线程数量
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
// 获取锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 完成任务数统计
completedTaskCount += w.completedTasks;
// 集合中移除工作线程
workers.remove(w);
} finally {
mainLock.unlock();
}
// 有worker线程移除，可能是最后一个线程退出，需要尝试终止线程池。
tryTerminate();
int c = ctl.get();
// RUNNGING 和 SHUTDOWN
if (runStateLessThan(c, STOP)) {
// 正常退出，计算需要维护的最小线程数
if (!completedAbruptly) {
// 核心线程，是否超时设置。默认是否，0说明最小线程数可为0；
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
// 确保当阻塞队列不为空时，至少有1个线程来处理剩余任务。
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 当前工作线程数量满足最小要求，返回。
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 不是正常退出，则添加一个非核心线程。用来处理剩余任务
addWorker(null, false);
}
}
通过completedAbruptly判断是否是正常退出。如果是正常退出的则不需要进行线程数量减法操作，因为在getTask中已经对线程数量做减法了；如果是非正常退出，则需要对线程数量做减法，然后在后续的逻辑判断中根据实际情况看是否还需要再创建新线程。无论是否是正常退出，都需要把当前工作线程从工作集合中移出。然后尝试终止线程池，因为有可能是最后一个工作线程退出。
最后，检查线程池的状态如果是RUNNGING或SHUTDOWN，则需要执行兜底补偿措施。确保能有工作线程来处理阻塞队列中剩余的任务。
8. tryTerminate()尝试关闭线程池
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// RUNNGING
// TIDYING 或 TERMINATED
// SHUTDOWN 且 队列不为空
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 到这里只有2种情况：STOP，SHUTDOWN & 队列为空
// 如果线程池中还有工作线程，则尝试中断
if (workerCountOf(c) != 0) { // Eligible to terminate
// STOP和线程池内还有工作线程，但是任务已经没有了。需要中断等待任务的线程。
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// CAS设置线程池为TIDYING状态
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// 终止线程池
terminated();
} finally {
// 线程池状态设置为 TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
当线程池涉及到要移除worker工作线程时都会调用此方法来判断是否需要结束当前线程池。只有STOP和SHUTDOWN队列为空时才会尝试执行结束，并且要保证工作线程数为0。通过CAS首先改变线程池状态到TIDYING，然后调用terminated()方法来结束线程池，此方法可以拓展。最后将线程池状态改为TERMINATED。
9. interruptIdleWorkers()中断空闲线程
// onlyONe 说明只终止一个线程，否则终止全部。
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
// 尝试对worker中断加锁。
if (!t.isInterrupted() && w.tryLock()) {
try {
// 对线程中断
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
对线程进行中断，对工作线程加锁。
