JAVA线程池源码全解析
线程池的具体使用方法和参数解析等我在之前已经讲解过，如果对线程池基本用法和概念不清晰的可以先看下我之前的线程池的文章，这里就通过一张线程池运行流程图来帮助大家去简单了解下线程池的工作原理。
线程池源码我们主要通过ThreadPoolExecutor进行分析，一步一步剖析线程池源码的核心内容。
属性解析
//高3位：表示当前线程池运行状态 除去高3位之后的低位：
// 表示当前线程池所拥有的线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 表示在ctl中，低COUNT_BITS位 用于存放当前线程数量的位
private static final int COUNT_BITS = Integer.SIZE - 3;
//低COUNT_BITS位 所能表达的最大数值
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;
// 所有任务已经中止，且工作线程数量为0，最后变迁到这个状态的线程将要执行terminated()钩子方法，只会有一个线程执行这个方法；
private static final int TIDYING = 2 << COUNT_BITS;
//中止状态，已经执行完terminated()钩子方法；
private static final int TERMINATED = 3 << COUNT_BITS;
//任务队列，当线程池中的线程达到核心线程数量时，再提交任务 就会直接提交到 workQueue
private final BlockingQueue<Runnable> workQueue;
//线程池全局锁，增加worker 减少 worker 时需要持有mainLock ， 修改线程池运行状态时，也需要。
private final ReentrantLock mainLock = new ReentrantLock();
//线程池中真正存放 worker->thread 的地方。
private final HashSet<Worker> workers = new HashSet<Worker>();
private final Condition termination = mainLock.newCondition();
// 记录线程池生命周期内 线程数最大值
private int largestPoolSize;
// 记录线程池所完成任务总数
private long completedTaskCount;
// 创建线程会使用线程工厂
private volatile ThreadFactory threadFactory;
/**
* 拒绝策略
*/
private volatile RejectedExecutionHandler handler;
//空闲线程存活时间，当allowCoreThreadTimeOut == false 时，会维护核心线程数量内的线程存活，超出部分会被超时。
//allowCoreThreadTimeOut == true 核心数量内的线程 空闲时 也会被回收。
private volatile long keepAliveTime;
//控制核心线程数量内的线程 是否可以被回收。true 可以，false不可以。
private volatile boolean allowCoreThreadTimeOut;
// 核心线程池数量
private volatile int corePoolSize;
// 线程池最大数量
private volatile int maximumPoolSize;
描述线程池状态的属性是贯穿整个线程池源码的核心，这里用一张图来描述一下。
running状态：当线程池是运行状态时，可以接收任务，也可以运行任务。
shutdown状态：此状态下，线程池不会再接收新的任务，当前的任务会继续执行完成
stop状态：当调用shutdownNow方法时，线程池会进入stop状态，不会接受新的任务，正在运行的任务也会被立即终止
tidying状态：进入该状态下此时线程池中任务和线程数量都为空
线程池运行任务可以通过submit方法和execute方法来完成
它量的区别在于submit会有返回值，返回Future对象，通过这个对象可以获取线程执行结果，execute没有返回值，下面来分别进行进行分析
execute方法
首先来分析execute方法，这也是线程池最核心的方法，因为submit方法其底层也是调用execute方法进行执行。
说execute方法之前，先来看下ThreadPoolExecutor的静态内部类Worker类。
//Worker采用了AQS的独占模式
//独占模式：两个重要属性 state 和 ExclusiveOwnerThread
//state：0时表示未被占用 > 0时表示被占用 < 0 时 表示初始状态，这种情况下不能被抢锁。
//ExclusiveOwnerThread:表示独占锁的线程。
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;
// worker内部封装的工作线程
final Thread thread;
//假设firstTask不为空，那么当worker启动后（内部的线程启动)会优先执行firstTask，当执行完firstTask后，会到queue中去获取下一个任务。
Runnable firstTask;
// 记录当前worker所完成的任务数量
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
*
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
// 设置AQS独占模式为初始化中的状态，这时候不能被抢占
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 使用线程工厂创建一个线程
this.thread = getThreadFactory().newThread(this);
}
线程池中的工作线程以Worker作为体现，真正工作的线程为Worker的成员变量，Worker即是Runnable，又是同步器。Worker从工作队列中取出任务来执行，并能通过Worker控制任务状态。
接下来通过execute方法源码来看下如何通过Worker完成任务的创建及运行。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 获取ctl的值
int c = ctl.get();
// 当前线程数小于核心线程池数量，此次提交任务，直接创建一个新的worker
// 相对应线程池多了一个新的线程
if (workerCountOf(c) < corePoolSize) {
// addWorker 即为创建线程的过程，会创建worker对象，并且将command作为firstTask
// core==true 表示采用核心线程数量限制，false采用maxinumPoolSize
if (addWorker(command, true))
return;
c = ctl.get();
}
// 执行到这里有几种情况？
// 1.当前线程池数量已经达到corePoolSize
// 2. addWorker失败
// 当前线程池处于running状态,尝试将task放入到workQueue中
if (isRunning(c) && workQueue.offer(command)) {
// 获取dangqctl
int recheck = ctl.get();
// !isRunning()成功，代表当你提交到任务队列后，线程池状态被外部线程给修改，例如调用了shutDown(),shutDownNow()
// remove成功，提交之后，线程池中的线程还没消费
// remove 失败，说明在shutDown或者shutDown之前,就被线程池的线程给处理了
if (!isRunning(recheck) && remove(command))
reject(command);
// 当前线程池是running状态，
else if (workerCountOf(recheck) == 0)
// 如果当前没有线程，就添加一个线程保证当前至少有一个线程存在
addWorker(null, false);
}
//执行到这里，有几种情况？
//1.offer失败
//2.当前线程池是非running状态
//1.offer失败，需要做什么？ 说明当前queue 满了！这个时候 如果当前线程数量尚未达到maximumPoolSize的话，会创建新的worker直接执行command
//假设当前线程数量达到maximumPoolSize的话，这里也会失败，也走拒绝策略。
//2.线程池状态为非running状态，这个时候因为 command != null addWorker 一定是返回false。
else if (!addWorker(command, false))
reject(command);
}
execute方法的执行流程大致可以分为以下几步：
工作线程数量小于核心数量，创建核心线程；
达到核心数量，进入任务队列；
任务队列满了，创建非核心线程；
达到最大数量，执行拒绝策略；
通过这个运行图再结合上面的源码可能对这个execute方法的具体执行流程就更加清楚了，下面就深入到每一个流程的细节去分析。
如果工作线程小于核心线程就会通过addWorker方法创建新的核心任务线程。
addWorker方法
//firstTask 可以为null，表示启动worker之后，worker自动到queue中获取任务.. 如果不是null，则worker优先执行firstTask
//core 采用的线程数限制 如果为true 采用 核心线程数限制 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.
// 判断当前线程池是否允许添加线程
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将线程数量加1，能够成功加1相当于申请到创建线程的令牌
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
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 创建work
w = new Worker(firstTask);
//将新创建的work节点的线程 赋值给t
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//持有全局锁，可能会阻塞，直到获取成功为止，同一时刻操纵 线程池内部相关的操作，都必须持锁。
mainLock.lock();
try {
//获取最新线程池运行状态保存到rs中
int rs = runStateOf(ctl.get());
//
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//将创建的work添加到线程池中
workers.add(w);
// 获取最新当前线程池线程数量
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
// 释放锁
mainLock.unlock();
}
if (workerAdded) {
// 添加work成功后，将创建的线程启动
t.start();
workerStarted = true;
}
}
} finally {
// 启动失败
if (!workerStarted)
// 释放令牌
// 将当前worker清理出workers集合
addWorkerFailed(w);
}
return workerStarted;
}
addWorker方法总体就是做了两件事
第一步:判断是否可以创建新的Work
第二步：如果可以创建就创建新的Work,然后添加到任务队列当中，最后启动该线程。
这里会看到，创建Work会加锁，加了一个来保证线程安全，新创建的Work会添加到任务队列当中，这个任务队列其实就是通过HashSet来存储work,最后启动线程，启动线程后，真正运行这个任务的方法就不在execute当中，而是通过
Work类中的run方法来执行。
runWorker方法
通过execute方法来启动线程后，就会通过work类中的run方法调用ThreadPoolExecutor的runWork方法来运行任务。
// 当worker启动时，会执行run方法
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
// 工作线程
Thread wt = Thread.currentThread();
// 任务
Runnable task = w.firstTask;
// 强制释放锁
// 这里相当于无视那边的中断标记
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 取任务，如果有第一个任务，这里先执行第一个任务
// 只要能取到任务，这就是个死循环
// getTask:取任务
while (task != null || (task = getTask()) != null) {
// 加锁，是因为当调用shutDown方法它会判断当前是否加锁，加锁就会跳过它接着执行下一个任务
w.lock();
// 检查线程池状态
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 钩子方法，方便子类在任务执行前做一些处理
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 真正任务执行的地方
//task 可能是FutureTask 也可能是 普通的Runnable接口实现类。
//如果前面是通过submit()提交的 runnable/callable 会被封装成 FutureTask。这个不清楚，请看上一期，在b站。
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方法就是真正执行任务的方法，如果有第一个任务就先执行第一个任务，第一个任务执行完后就通过getTask（）方法从任务队列中获取任务来执行。
getTask()方法
private Runnable getTask() {
// 是否超时
boolean timedOut = false; // Did the last poll() time out?
// 自旋
for (; ; ) {
int c = ctl.get();
int rs = runStateOf(c);
//当前程池状态是SHUTDOWN的时候会把队列中的任务执行完直到队列为空
// 线程池状态是stop时退出
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
// 获取工作线程数量
int wc = workerCountOf(c);
// 是否允许超时，有两种情况：
// 1. 是允许核心线程数超时，这种就是说所有的线程都可能超时
// 2. 是工作线程数大于了核心数量，这种肯定是允许超时的
// 注意，非核心线程是一定允许超时的，这里的超时其实是指取任务超时
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 真正取任务的地方
// 默认情况，只有当工作线程数量大于核心线程数量时，才会调用poll方法触发超时调用
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
// 取到任务就返回
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
这里取任务会根据工作线程的数量判断是使用BlockingQueue的poll(timeout, unit)方法还是take()方法。
poll(timeout, unit)方法会在超时时返回null，如果timeout<=0，队列为空时直接返回null。
take()方法会一直阻塞直到取到任务或抛出中断异常。
submit方法
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
submit方法是支持传入Runnable或者Callable,通过newFaskFor方法将其包装到FutureTask进行处理，FutureTask会在下篇文章进行详细讲解，futureTask主要做了两件事，一件事是扩展run方法，用来完成结果值的处理，另一件事是暴露其get方法，通过get方法获取执行结果，这个get方法是阻塞的。
我做了一张思维导图，通过这个思维导图来梳理一遍线程池的大概脉络
PS：如果觉得我的分享不错，欢迎大家随手点赞、在看。
（完） 加我"微信" 获取一份 最新Java面试题资料 请备注：666，不然不通过～
最近好文
1、Spring Boot 实现扫码登录，这种方式太香了！！
最近面试BAT，整理一份面试资料《Java面试BAT通关手册》，覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。 获取方式：关注公众号并回复 java 领取，更多内容陆续奉上。 明天见(｡･ω･｡)ﾉ♡
评论