线程池在工程中扮演着不可或缺的角色，话不多说，今天就来看看它底层的运行原理。
初始化参数
关于这个问题也经常被问到，线程池创建都有哪些参数，它们都有什么含义，我们先从构造参数开始入手
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
// 省略代码
}
- corePoolSize，核心线程数量，可以理解为这些线程是线程池的「中流砥柱」，线程池实现线程复用很大程度是依赖于这个核心线程数
- maximumPoolSize，最大线程数量，核心线程忙不过来的情况下，会额外创建非核心线程来分担任务，它用来限制线程池能够创建的有效线程数量
- keepAliveTime，超时时间，线程池发现某些非核心线程「游手好闲」太久了，认为它光领工资不干活，空闲时间到了keepAliveTime 之后会把它干掉，必要时候核心线程也会受这个超时时间影响，通过设置 allowCoreThreadTimeOut = true
- unit，keepAliveTime 的时间单位
- workQueue， 存放任务的队列，很多时候线程池没有办法马上就处理刚提交的任务，因此需要将其存放在任务队列中，等待后续处理
- handler，如果前面说的线程数量已经到达上限，并且都忙得不可开交，任务队列的任务也放满了，这个时候线程池会认为人手不够，会使用预先设置的拒绝策略拒绝掉后续提交的任务。
讲完了线程池的构造参数，相信你对线程池已经有了大体的认识，趁热打铁，接着往下。
主流程
先来看一下任务处理的主流程，找到 execute() 方法，这是我们提交任务常用的方法。
- 如果当前线程的数量少于核心线程数，尝试创建一个核心线程来运行任务
- 如果上一步条件不成立，尝试把当前任务存放到任务队列，如果能够成功入队，还需要进行 double-check，防止这个时候线程池刚好被 shutdown 或者有线程挂掉了，必要时候会对已经入队的任务进行回滚，或是创建一条新的 worker 线程来处理已经入队的任务
- 如果上面条件都不成立，同时线程数量已经达到上限，会采用拒绝策略来处理任务
public void execute(Runnable command) {
// 省略代码
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))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
3.必要时候拒绝掉当前提交任务
else if (!addWorker(command, false))
reject(command);
}
流程图如下：
image
了解了任务处理的流程之后，接下来我们来看一些细节上的实现。
提交任务
从上面的流程可知，线程池很多时候需要创建工作线程来处理提交的任务，主要看到 addWorker() 这个方法，方法比较长，我们拆分成几部分来讲。
第一部分：前置条件判定
- 判断是否可以创建线程
- 判断当前线程数量是否超出限制
- 采用 CAS 的方式对线程数量进行 +1 操作
- 如果中途线程池状态发生改变，重新进行最外层的条件判断，是否需要创建线程
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
1.判断是否可以创建线程
条件1：线程池不是 RUNNING 状态
条件2-1：线程池状态为 STOP，TIDYING，TERMINATED中的一种，这时候意味着不再处理任务，结束创建线程操作
条件2-2：线程池状态为 SHUTDOWN，firstTask 参数不为空，此时线程池不再接受新任务，结束创建线程操作
条件2-3：线程池状态为 SHUTDOWN，firstTask 参数为空，如果任务队列也为空，结束创建线程操作
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
2.判断当前线程数量是否超出限制
如果创建核心线程，则和 corePoolSize 比较
如果创建非核心线程，则和 maximumPoolSize 比较
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
3.采用 CAS 的方式对线程数量进行 +1 操作
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
4.如果中途线程池状态发生改变，重新进行最外层的条件判断，是否需要创建线程
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
}
流程图如下：
image
第二部分：创建 worker 线程
- 构造 worker 对象
- 重新检查是否满足创建线程的要求
- 将 worker 对象加入列表中，更新线程池线程数量
- 开启 worker 线程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
1.构造 worker 对象
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
2.重新检查是否满足创建线程的要求
如果线程池不是 RUNNING 状态，则不创建线程
如果线程池为 SHUTDOWN 状态，并且 firstTask 不为空，此时不进行新任务的处理，不创建线程
如果线程池状态为 STOP，TIDYING，TERMINATED，也不创建线程
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
3.将 worker 对象加入列表中，更新线程池线程数量
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
4.开启 worker 线程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
流程图如下：
image
总结
通过这篇文章，我们了解了：
- 线程池的构造参数
- 线程池的任务提交流程
大家可能对线程池任务执行机制还留有疑问，考虑到篇幅有限，打算将这一部分拆分到下一次分享，如果内容对大家有帮助，记得点赞哦。
网友评论