聊聊 ThreadPoolExecutor 线程池及其源码

什么是线程池

在我们的实际开发中，通常使用线程池创建线程资源。

而线程池本质上是一种池化技术，利用资源复用的思想，复用线程资源。

为什么使用线程池

那么，我们为什么要使用线程池呢？主要有以下两点原因：

• 首先，使用线程池可以避免频繁创建和销毁线程所带来的性能开销。因为线程的创建会涉及到​​CPU​​ 上下文的切换以及内存的分配。
• 其次，我们可以通过线程池参数控制线程的数量，避免无休止的创建线程带来的资源利用率过高的问题，从而起到资源保护的作用。

线程池的状态

在我们使用线程池时，一共涉及 ​​5​​ 种状态，其中包括：

​​RUNNING​​ ：能接受新提交的任务，并且也能处理阻塞队列中的任务；

​​SHUTDOWN​​：关闭状态，不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务。在线程池处于 RUNNING 状态时，调用 shutdown() 方法会使线程池进入到该状态。（finalize() 方法在执行过程中也会调用 shutdown() 方法进入该状态）；

​​STOP​​：不能接受新任务，也不处理队列中的任务，会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时，调用 shutdownNow() 方法会使线程池进入到该状态；

​​TIDYING​​：如果所有的任务都已终止了，workerCount (有效线程数) 为 0，线程池进入该状态后会调用 terminated() 方法进入 TERMINATED 状态。

​TERMINATED​​：在 terminated() 方法执行完后进入该状态，默认 terminated() 方法中什么也没有做。进入TERMINATED的条件如下：

• 线程池不是​​RUNNING​​ 状态；
• 线程池状态不是​​TIDYING​​ 状态或​​TERMINATED​​ 状态；
• 如果线程池状态是​​SHUTDOWN​​ 并且​​workerQueue​​ 为空；
• ​​workerCount​​ 为​​0​​；
• 设置​​TIDYING​​ 状态成功。

线程池的分析及源码

核心属性

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS = 32 - 3 = 29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 工作线程的最大个数
// 00100000 00000000 00000000 00000000 - 1
// 000111111111111111111111111111111
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;
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​​ 属性维护着线程池的两个重要内容：

高​​3​​ 位维护着线程池的状态；

低​​29​​ 位维护着工作线程个数；

ctl 相关方法

// 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; }

ThreadPoolExecutor 构造方法

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}

可以看出，在 ​​ThreadPoolExecutor​​ 的有参构造器中一共有 ​​6​​ 个参数：

• ​​corePoolSize​​：核心线程数
• ​​maximumPoolSize​​：最大线程数
• ​​keepAliveTime​​：线程池维护线程所允许的空闲时间，当线程池中的线程数量大于 corePoolSize 的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了 keepAliveTime
• ​​unit​​：时间单位
• ​​workQueue​​：等待队列，当任务提交时，如果线程池中的线程数量大于等于 corePoolSize 的时候，把该任务封装成一个 Worker 对象放入等待队列
• ​​threadFactory​​：它是 ThreadFactory 类型的变量，线程工厂，用来创建新的线程
• ​handler​​：它是 RejectedExecutionHandler 类型的变量，表示线程池的饱和策略。如果阻塞队列满了并且没有空闲的线程，这时如果继续提交任务，就需要采取一种策略处理该任务。线程池提供了 4 种策略：

​​AbortPolicy​​：直接抛出异常，这是默认策略；

​​CallerRunsPolicy​​：用调用者所在的线程来执行任务；

​​DiscardOldestPolicy​​：丢弃阻塞队列中靠最前的任务，并执行当前任务；

​​DiscardPolicy​​：直接丢弃任务；

核心思想

当有新任务提交的时候，首先会执行以下判断：

如果运行的线程少于​​corePoolSize​​，则创建新线程来处理任务，即使线程池中的其他线程是空闲的；

如果线程池中的线程数量大于等于​​corePoolSize​​ 且小于​​maximumPoolSize​​，则只有当​​workQueue​​ 满时才创建新的线程去处理任务；

如果设置的​​corePoolSize​​ 和​​maximumPoolSize​​ 相同，则创建的线程池的大小是固定的，这时如果有新任务提交，若​​workQueue​​ 未满，则将请求放入​​workQueue​​ 中，等待有空闲的线程去从​​workQueue​​ 中取任务并处理；

如果运行的线程数量大于等于​​maximumPoolSize​​，这时如果​​workQueue​​ 已经满了，则通过​​handler​​ 所指定的策略来处理任务；

所以，任务提交时，判断的顺序为 ​​corePoolSize​​ –> ​​workQueue​​ –> ​​maximumPoolSize​​。

源码

execute 方法

public void execute(Runnable command) {
// 首先判断任务是否为null，如果为null，抛出空指针异常
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1\. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2\. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3\. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
// 获取ctl
int c = ctl.get();
// workerCountOf方法是获取工作线程数
// 如果当前工作线程数小于corePoolSize核心线程数
if (workerCountOf(c) < corePoolSize) {
// 添加核心线程，其中true是根据corePoolSize判断，false是根据maximumPoolSize判断
if (addWorker(command, true))
// 如果添加核心线程成功，直接return
return;
// 如果失败，再次获取ctl
c = ctl.get();
}
// 如果当前线程池状态是Running，并且任务添加到workQueue队列中成功
if (isRunning(c) && workQueue.offer(command)) {
// 再次获取ctl
int recheck = ctl.get();
// 再次判断线程池状态，如果线程池状态不是Running了，需要将刚才添加到workQueue队列中的任务移除
if (! isRunning(recheck) && remove(command))
// 移除workQueue队列中的任务
reject(command);
// 如果工作线程数为0，则执行addWorker方法
// 第一个参数为null，表示在线程池中创建一个线程，但不去启动
// 第二个参数为false，代表是根据maximumPoolSize进行判断
// 如果判断workerCount大于0，则直接返回，在workQueue中新增的command会在将来的某个时刻被执行。
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果执行到这里，有两种情况：
// 1\. 线程池已经不是RUNNING状态；
// 2\. 线程池是RUNNING状态，但workerCount >= corePoolSize并且workQueue已满。
// 这时，再次调用addWorker方法，但第二个参数传入为false，将线程池的有限线程数量的上限设置为maximumPoolSize；
// 如果失败则拒绝该任务
else if (!addWorker(command, false))
reject(command);
}

我们在执行 ​​execute()​​ 方法时，如果线程池状态一直是 ​​RUNNING​​ 的话：

如果​​workerCount < corePoolSize​​，则创建并启动一个线程来执行新提交的任务；

如果​​workerCount >= corePoolSize​​，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中；

如果​​workerCount >= corePoolSize && workerCount < maximumPoolSize​​，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务；

如果​​workerCount >= maximumPoolSize​​，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。

这里要注意一下 ​​addWorker(null, false);​​，也就是创建一个线程，但并没有传入任务，因为任务已经被添加到workQueue中了，所以 ​​worker​​ 在执行的时候，会直接从 ​​workQueue​​ 中获取任务。所以，在 ​​workerCountOf(recheck) == 0​​ 时执行 ​​addWorker(null, false);​​ 也是为了保证线程池在 ​​RUNNING​​ 状态下必须要有一个线程来执行任务。

addWorker 方法

​​addWorker()​​ 方法的主要工作是在线程池中创建一个新的线程并执行。其中 ​​firstTask​​ 参数为新创建线程第一个需要执行的任务。​​core​​ 参数如果为 ​​true​​，则使用 ​​corePoolSize​​ 进行判断，​​false​​ 则使用 ​​maximumPoolSize​​。

private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
// 获取ctl
int c = ctl.get();
// 获取线程池状态
int rs = runStateOf(c);

// Check if queue empty only if necessary.
// 首先，如果rs >= SHUTDOWN，则表示线程池不再接收新任务
// 之后，再判断后面三个条件，只要有一个不满足，就返回false
// 如果rs == SHUTDOWN，因为不再接收任务了，firstTask不是null，返回false
// 如果firstTask是null，并且workQueue为空，也返回false
// 因为workQueue已经为空了，不需要再创建线程了
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;

for (;;) {
// 获取工作线程
int wc = workerCountOf(c);
// 首先判断wc >= CAPACITY，如果大于等于工作线程最大值，返回false
// 之后，如果core是true，则与corePoolSize比较，是false，则与maximumPoolSize比较
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 对工作线程执行+1操作，
if (compareAndIncrementWorkerCount(c))
// 如果+1成功，跳出外层循环
// CAS方式，并发操作，只能有一个线程+1成功
break retry;
c = ctl.get(); // Re-read ctl
// 判断线程池状态是否有变化
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// 工作线程是否启动标识
boolean workerStarted = false;
// 工作线程是否添加表示标识
boolean workerAdded = false;
// Worker就是工作线程
Worker w = null;
try {
// 创建新的工作线程，将firstTask传入Worker对象
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());
// 如果线程池是RUNNING状态，或者线程池是SHUTDOWN状态且firstTask是null
// 向线程池中添加线程
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// workers是HashSet，保存线程池中的所有工作线程
workers.add(w);
int s = workers.size();
// 记录线程池中出现的最大线程数
if (s > largestPoolSize)
largestPoolSize = s;
// 将线程添加状态置为true，代表线程添加成功
workerAdded = true;
}
} finally {
// 释放锁
mainLock.unlock();
}
// 如果线程添加成功，启动线程，将线程启动状态置为true
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

Worker 类

线程池中，每一个线程被封装成一个 ​​Worker​​ 对象，​​ThreadPool​​ 维护的其实就是一组 ​​Worker​​ 对象。

我们可以看到 ​​Worker​​ 类继承了 ​​AQS​​，实现了 ​​Runnable​​ 接口。其中包括两个属性，​​firstTask​​ 和 ​​thread​​。其中 ​​firstTask​​ 用来保存传入的任务，而 ​​thread​​ 是在调用构造方法时通过 ​​ThreadFactory​​ 类来创建的线程，用来处理任务。

private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;

/** 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;

/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}

/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}

// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.

protected boolean isHeldExclusively() {
return getState() != 0;
}

protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}

protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}

public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }

void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}

runWorker 方法

因为 ​​Worker​​ 类实现了 ​​Runnable​​ 接口，因此，在 ​​Worker​​ 对象启动的时候，会调用重新的 ​​run​​ 方法，​​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 {
// 如果task为null，则通过getTask来获取任务
// 也就是Worker自身有携带任务，直接执行
// 如果Worker自身没有携带任务，那么就通过getTask方法取工作队列中获取任务
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 判断线程池状态是否大于等于STOP，如果是，要中断当前线程
if ((runStateAtLeast(ctl.get(), STOP) ||
// DCL
(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);
}
}

​​runWorker​​ 方法的执行过程：

​​while​​ 循环不断地通过​​getTask()​​ 方法获取任务；

​​getTask()​​ 方法从阻塞队列中取任务；

如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态；

调用​​task.run()​​ 方法执行任务；

如果​​task​​ 为​​null​​ 则跳出循环，执行​​processWorkerExit()​​ 方法；

​​runWork()​​ 方法执行完毕，也代表着​​Worker​​ 中的​​run​​ 方法执行完毕，摧毁线程；

getTask 方法

​​getTask​​ 方法主要用来在阻塞队列中获取任务。

private Runnable getTask() {
// timeOut变量的值表示上次从阻塞队列中取任务时是否超时
boolean timedOut = false; // Did the last poll() time out?

for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// Check if queue empty only if necessary.
// 如果线程池状态rs >= SHUTDOWN，也就是说是非RUNNING状态，那么再判断
// rs >= STOP，线程池是否正在stop
// 阻塞队列为空
// 如果上面两个条件有一个满足，需要工作线程数-1，并返回null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}

int wc = workerCountOf(c);

// Are workers subject to culling?
// 判断核心线程是否允许超时
// 判断工作线程数时候大于核心线程数
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 判断工作线程是否超过最大线程数
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
// 工作线程数-1
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}

try {
// 如果是非核心，走poll，拉取工作队列任务
// 如果是核心线程，走take一直阻塞，拉取工作队列任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}

processWorkerExit 方法

private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果completedAbruptly为true，说明线程在执行时出现了异常，需要对工作线程数减1
// 执行正常的情况下，是在getTask方法中进行减1操作
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 将当前工作线程完成的任务个数赋值给整个线程池中的任务数
completedTaskCount += w.completedTasks;
// 从workers中移除，表示从线程池中移除了一个工作线程
workers.remove(w);
} finally {
mainLock.unlock();
}
// 判断线程池是否可以中止，线程池状态是否发生变化
tryTerminate();

int c = ctl.get();
// 如果当前线程池状态小于STOP
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);
}
}