一、引言
一直以来都想写一个关于线程池源码的分析，但是这里面涉及的知识点很多没办法一次写完，因此一拖再多，现在实在不想再拖延下去了，总结一下线程池一些问题。如有不当的地方请各位同行及时指出，谢谢各位。
二、线程池的参数介绍
java.uitl.concurrent.ThreadPoolExecutor类就是我们构造线程池的核心类，首先看一下这个类的构造。
1、ThreadPoolExecutor的UML图
从UML图中我们可以看出ThreadPoolExecutor继承自抽象类AbstractExecutorService实现ExecutorService接口继承自Executor接口 。
public class ThreadPoolExecutor extends AbstractExecutorService {
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), handler);
}
//最终调用的构造方法
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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
}
从上面的构造方法我们可以看出其他的构造方法都是调用最后一个构造方法完成初始化的。从上到下解释一下参数的含义：
- int corePoolSize：线程池核心线程数。根据这个英文的翻译我们可以看出这个值是线程池的核心数。默认情况下线程池创建的时候并没有创建任何线程，而是等到有任务的时候才创建线程去执行任务。除非调用了除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。当corePoolSize线程来不及处理task任务的时候，任务就存到workQueue阻塞队列。
- int maximumPoolSize：线程池的最大线程数。它标识线程池最多容纳的线程数，如果核心线程已满，并且workQueue阻塞队列也满了线程池将会扩容到maximumPoolSize最大线程数处理task任务。
- long keepAliveTime：表示线程空闲多长时间将会终止。默认情况下，只有当线程池数量大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数恢复到不大于corePoolSize。但是如果调用了allowCoreThreadTimeOut(true)方法，根据方法名我们也可以理解，如果调用这个方法就是允许核心线程超时，因此线程池线程数可减少到0。
- TimeUnit unit：keepAliveTime的单位。 TimeUnit.DAYS; //天 TimeUnit.HOURS; //小时 TimeUnit.MINUTES; //分钟 TimeUnit.SECONDS; //秒 TimeUnit.MILLISECONDS; //毫秒 TimeUnit.MICROSECONDS; //微妙 TimeUnit.NANOSECONDS; //纳秒
- BlockingQueue<Runnable> workQueue：存放task的阻塞队列。当核心线程来不及处理的任务就会存到这个阻塞队列中。常用的有这几种选择：
- ArrayBlockingQueue：是有界队列，内部实现是将对象放到一个数组里。遵循FIFO先进先出原则。生产一般都使用有界队列配合线程池使用。
- LinkedBlockingQueue：是无界队列，内部实现是 以链式结构对元素进行存储。遵循FIFO先进先出原则。使用该队列作为阻塞队列的时候要留意，如果没有设置上限，将会用Integer.MAX_VALUE 作为上限。
- SynchronousQueue：同步移交队列，内部只够容纳单个元素。每个操作必须等到两个线程调用溢出操作，否则一直处于阻塞的状态。
- priorityBlockingQuene：具有优先级的无界阻塞队列。所有插入到 priorityBlockingQuene的元素必须实现 java.lang.Comparable 接口。因此该队列中元素的排序就取决于你自己的 Comparable 实现。注意： (1) LinkedBlockingQueue比ArrayBlockingQueue在插入删除节点性能方面更优，因此有更大的吞吐量。但是二者在put(), take()任务的时均需要加锁。 (2) SynchronousQueue使用无锁算法，根据节点的状态判断执行，而不需要用到锁，其核心是Transfer.transfer()
- ThreadFactory threadFactory：线程工厂，主要用来创建worker线程。
- RejectedExecutionHandler handler：处理任务的策略，有以下取值：
- ThreadPoolExecutor.AbortPolicy（默认策略）:丢弃任务并抛RejectedExecutionException异常。
- ThreadPoolExecutor.DiscardPolicy：丢弃任务，但是不抛出异常。
- ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行当前任务。
- ThreadPoolExecutor.CallerRunsPolicy：调用者所在的线程来处理该任务。 也可以根据应用使用场景来实现RejectedExecutionHandler接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。
三、线程池的使用
1、线程池的工作原理
简要说明一下线程池工作原理：
- 使用者提交新的task任务后线程池首先判断核心线程池是否已满，如果未满则重新创建新的worker线程。否则进入2。
- 判断任务队列是否已经满了，如果还没满将任务放入队列。否则进入3。
- 判断线程池的线程是否有空闲，如果没有，则重新创建worker线程。如果线程池已满，则执行拒绝策略。
- 如果worker线程处于空闲状态，则执行线程超时销毁。
线程池工作原理图如下:
线程池工作原理图：https://www.processon.com/view/link/5ef2d5545653bb2925b2ebeb
2、线程池类型
Java通过Executors（jdk1.5并发包）提供四种线程池，分别为：
- newCachedThreadPool：创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
- newFixedThreadPool ：创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
- newScheduledThreadPool ：创建一个定长线程池，支持定时及周期性任务执行。
- newSingleThreadExecutor： 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
2.1、newCachedThreadPool使用
通过构造方法源码，我们可以看出这个线程池核心线程是0的无界线程池，也就是说空闲时间足够长核心线程可以变成0不消耗系统资源。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
测试代码如下：
public static void newCachedThreadPoolTest() throws InterruptedException {
//创建可缓存的线程池
ExecutorService threadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int finalI = i;
threadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + ",我是任务：" + finalI);
}
});
}
//休息5秒关闭线程池
TimeUnit.SECONDS.sleep(5);
threadPool.shutdown();
}
打印结果：
pool-1-thread-1,我是任务：0
pool-1-thread-2,我是任务：1
pool-1-thread-3,我是任务：2
pool-1-thread-4,我是任务：3
pool-1-thread-6,我是任务：5
pool-1-thread-5,我是任务：4
pool-1-thread-7,我是任务：6
pool-1-thread-8,我是任务：7
pool-1-thread-9,我是任务：8
pool-1-thread-10,我是任务：9
2.2、newFixedThreadPool使用
通过构造方法我们可以看出这是个可定容量的线程池，大于核心线程的工作线程用完立即销毁。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
测试代码如下：
public static void newFixedThreadPoolTest() throws InterruptedException {
//创建定长的线程池
ExecutorService threadPool = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
final int finalI = i;
threadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + ",我是任务：" + finalI);
}
});
}
//休息5秒关闭线程池
TimeUnit.SECONDS.sleep(5);
threadPool.shutdown();
}
打印结果：
pool-1-thread-1,我是任务：0
pool-1-thread-3,我是任务：2
pool-1-thread-3,我是任务：6
pool-1-thread-5,我是任务：4
pool-1-thread-5,我是任务：8
pool-1-thread-1,我是任务：5
pool-1-thread-5,我是任务：9
pool-1-thread-3,我是任务：7
pool-1-thread-4,我是任务：3
pool-1-thread-2,我是任务：1
通过结果我们可以看出，线程池只有5个线程。
2.3、newScheduledThreadPool使用
通过构造方法我们可以看出这是个可定容量的线程池，他这里有schedule方法可以延迟线程池执行。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
测试代码：
public static void newScheduledThreadPoolTest() throws InterruptedException {
//定时线程池
ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(5);
for (int i = 0; i < 10; i++) {
final int finalI = i;
threadPool.schedule(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + ",我是任务：" + finalI);
}
}, 2, TimeUnit.SECONDS);//等待2秒执行
}
//休息5秒关闭线程池
TimeUnit.SECONDS.sleep(5);
threadPool.shutdown();
}
打印结果：
pool-1-thread-5,我是任务：4
pool-1-thread-1,我是任务：0
pool-1-thread-4,我是任务：3
pool-1-thread-3,我是任务：2
pool-1-thread-2,我是任务：1
pool-1-thread-3,我是任务：8
pool-1-thread-4,我是任务：7
pool-1-thread-1,我是任务：6
pool-1-thread-5,我是任务：5
pool-1-thread-2,我是任务：9
2.4、newSingleThreadExecutor使用
通过构造方法我们可以看出这是个线程池只有一个工作线程。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
测试代码
public static void newSingleThreadExecutorTest() throws InterruptedException {
//创建定长的线程池
ExecutorService threadPool = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int finalI = i;
threadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + ",我是任务：" + finalI);
}
});
}
//休息5秒关闭线程池
TimeUnit.SECONDS.sleep(5);
threadPool.shutdown();
}
打印日志：
pool-1-thread-1,我是任务：0
pool-1-thread-1,我是任务：1
pool-1-thread-1,我是任务：2
pool-1-thread-1,我是任务：3
pool-1-thread-1,我是任务：4
pool-1-thread-1,我是任务：5
pool-1-thread-1,我是任务：6
pool-1-thread-1,我是任务：7
pool-1-thread-1,我是任务：8
pool-1-thread-1,我是任务：9
四、线程池的源码实现
1、线程池的状态
主线程池控制状态ctl是一个原子整数变量，包含两个部分：
- runState：线程的状态运行、关闭状态等，使用高3位表示（因为3位可以表示8个数，线程池状态只有5个足够了）
- workerCount：有效线程数量，使用第29位表示，因此线程数的最多为2^29 -1
/**
* 主线程池控制状态ctl是一个原子整数变量包含两个概念字段：
* workerCount：有效线程数量
* runState：线程的状态运行、关闭状态等。
* <p/>
* 状态转换的几种情况
* RUNNING -> SHUTDOWN 执行了shutdown()方法或者间接的执行了finalize()方法
* (RUNNING or SHUTDOWN) -> STOP 执行了shutdownNow()方法
* SHUTDOWN -> TIDYING 当队列和线程池都是空的时候
* TIDYING -> TERMINATED 当terminated()钩子方法完成时
* <p/>
* 等待awaitTermination()的线程将在状态达到TERMINATED时return
*/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
/**
* 低29位，COUNT_BITS就是29
*/
private static final int COUNT_BITS = Integer.SIZE - 3;
/**
* 这个就是workerCount线程数量最大值，其实就是2^29 -1
*/
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
/**
* 使用高3位就可以表示全部状态了，因为状态一共才5个，在计算机中数字都是补码表示，
* 下面的移位操作都是按照补码来移位
* <p/>
* RUNNING: 接受新任务并处理排队的任务，并且也能处理阻塞队列中
* 111.....0
*/
private static final int RUNNING = -1 << COUNT_BITS;
/**
* SHUTDOWN: 关闭状态，不再接受新提交的任务，但可以继续处理阻塞队列中已保存的任务
* 000......0
*/
private static final int SHUTDOWN = 0 << COUNT_BITS;
/**
* STOP: 不接受新任务、不处理排队的任务，会中断正在处理任务的线程
* 001......0
*/
private static final int STOP = 1 << COUNT_BITS;
/**
* TIDYING: 所有任务都已终止，workerCount为零
* 010......0
*/
private static final int TIDYING = 2 << COUNT_BITS;
/**
* TERMINATED: terminated()钩子方法执行完成后进入该状态
* 011......0
*/
private static final int TERMINATED = 3 << COUNT_BITS;
2、ThreadPoolExecutor重要的成员变量
线程池中以下成员变量比较重要，先要解释一下。
private volatile int corePoolSize; //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int maximumPoolSize; //线程池最大能容纳的线程数
private final BlockingQueue<Runnable> workQueue; //任务缓存队列，用来存放等待执行的任务
private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间
private volatile long keepAliveTime; //线程存活时间
private volatile int poolSize; //线程池中当前的线程数
private volatile RejectedExecutionHandler handler; //任务拒绝策略
private final ReentrantLock mainLock = new ReentrantLock(); //线程池内部独占锁，对线程池统计信息（线程池大小、runState等）的改变都使用这个锁
private final HashSet<Worker> workers = new HashSet<Worker>();//内部运行的Worker线程存放的地方，通过mainLock保证线程安全
private volatile ThreadFactory threadFactory; //线程工厂，用来创建线程
private int largestPoolSize; //用来记录线程池中曾经出现过的最大线程数，统计信息
private long completedTaskCount; //用来记录已经执行完毕的任务个数
3、线程池中任务提交
ThreadPoolExecutor实现的实际是一个生产者-消费者模型，添加任务相当于生产者生产元素，workers线程工作集中的线程直接执行任务或者从任务队列中取出任务相当于消费者消费元素。提交任务方法execute源码如下：
public void execute(Runnable command) {
//(1)如果任务为null，则直接抛出NPE异常
if (command == null)
throw new NullPointerException();
//(2)获取当前线程池的状态+线程个数变量的控制器ctl
int c = ctl.get();
//(3)当前线程个数是否小于corePoolSize,小于则添加一个新worker线程运行
if (workerCountOf(c) < corePoolSize) {
//(3.1)如果核心线程添加成功这返回
if (addWorker(command, true))
return;
//(3.2)如果到这里代表核心线程添加失败了，则重新获取ctl的值
c = ctl.get();
}
/**
* 添加核心worker线程失败的几种场景：
* 1.线程池为SHUTDOWN以上的状态
* 2.当前线程池这种运行的worker的数量超过最大限制（2^29-1）
* 3.当前线程池中运行的worker的数量超过corePoolSize
*/
//(4)如果线程池处于RUNNING状态，则把任务添加到阻塞队列
if (isRunning(c) && workQueue.offer(command)) {
//(4.1)二次检查线程池是否处于RUNNING状态
int recheck = ctl.get();
//(4.2)如果当前线程池不是RUNNING状态则将任务从队列中移除并且执行拒绝策略
if (!isRunning(recheck) && remove(command))
reject(command);
//(4.3)否则，如果当前的线程池为空，这添加一个worker线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//(5)如果队列满了，则新增线程，新增失败则执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
其中的addWorker方法源码如下：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (; ; ) {
//获取当前线程池的状态+线程个数变量的控制器ctl
int c = ctl.get();
//获取线程池的状态
int rs = runStateOf(c);
/**
* 提交任务过程中，如果此时线程池执行了shutdown操作，则不创建新线程
* 第一个条件：rs >= SHUTDOWN，也就是此时已经执行了shutdown操作，可能的状态为有SHUTDOWN、STOP、TIDYING、TERMINATED
* 第二个条件：!(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty())
* 等价于(!(rs == SHUTDOWN) || !(firstTask == null) || workQueue.isEmpty())只要有个一个为true就成立
* 1.!(rs == SHUTDOWN) 为true表示rs!=SHUTDOWN的状态，也就是线程时处于SHUTDOWN、STOP、TIDYING、TERMINATED,不需要创建worker线程了
* 2.(firstTask != null) 为true表示提交任务过程中，线程池刚好处于SHUTDOWN状态，不需要创建worker线程了
* 3.workQueue.isEmpty() 为true表示任务队列已经空了，不需要创建worker线程了
*/
//(6)检查队列是否是SHUTDOWN以上的状态
if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
return false;
//(7)循环CAS增加线程个数
for (; ; ) {
int wc = workerCountOf(c);//获取当前的工作线程个数
//(7.1)如果线程个数超过限制则返回false，即添加工作线程失败
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//(7.2)CAS操作workCount+1,如果+1成功则跳出循环
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();//重新读取ctl的值
//(7.3)CAS操作失败了，看线程池状态是否发生变化了，变化了则跳到外层循环重新获取线程池状态，否则内存循环重新获取CAS
if (runStateOf(c) != rs)
continue retry;
}
}
//(8)能到这里说明CAS成功了
boolean workerStarted = false;//worker线程是否启动成功标志
boolean workerAdded = false;//是否已经将worker线程添加到workers这个HashSet<Worker>集合中
Worker w = null;
try {
//(8.1)创建worker线程
w = new Worker(firstTask);//定义一个worker线程，这个woker线程绑定的是提交的task任务
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//(8.2)加独占锁，为了workers同步，因为可能多个线程调用了线程池的execute()方法
mainLock.lock();
try {
//(8.3)重新检查线程池状态，为了避免在ThreadFactory失败或者在获取锁之前调用了shutdown方法
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive())//预先检查t是否可启动
throw new IllegalThreadStateException();
//(8.4)将worker线程添加到集合中
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)//largestPoolSize记录最大活跃线程数
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//(8.5)添加成功则启动任务
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//(8.6)如果启动失败了，则回退操作
if (!workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
4、线程池中worker的执行
当我们完成了提交任务的过程的时候，下面一步就是worker线程是如何执行任务的。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
Worker(Runnable firstTask) {
setState(-1); //在线程执行runWorker之前禁止被中断
this.firstTask = firstTask;//外部提交的任务
this.thread = getThreadFactory().newThread(this);//真实的执行任务的线程
}
}
从Worker类中我们可以看出它是实现了Runnable接口，并且是AQS的子类，那么我们可以推测出它能够进行并发的控制（lock、unlock）。runWorker的实现如下：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();//在woker的流程中执行worker.start()之后真实调用的方法
Runnable task = w.firstTask;//获取当前worker携带的task任务
w.firstTask = null;
/**
* 这里直接调用了unlock方法，但是我们并没有看到调用lock方法，说明unlock之前不一定需要lock
*/
//(9)state设置成0，将占用锁的线程设置为null（第一次执行之前没有线程占用）
w.unlock();
boolean completedAbruptly = true;
try {
//(10.0)自旋。先执行自己携带的任务，然后从阻塞队列中获取一个任务直到无法获取任务
while (task != null || (task = getTask()) != null) {
//(10.1)将satte设置为1，设置占有锁的线程为自己
w.lock();
//(10.2)检查线程池状态，如果状态为STOP以上(STOP以上不执行任务),并且当前线程还未被中断则中断当前线程。
//第二次检查状态是为了处理shutdownNow操作
if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP)))
&& !wt.isInterrupted())
wt.interrupt();
try {
//(10.3)这是?个钩??法，留给需要的?类实现
beforeExecute(wt, task);
Throwable thrown = null;
try {
//(10.4)执?任务?法,若任务执?发?异常,当前worker不会再继续执?任务,线程销毁,后续会新增?个线程进?补偿
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 {
//(10.5)钩??法，执?任务处理后逻辑,如异常处理
afterExecute(task, thrown);
}
} finally {
task = null;
//(10.6)任务执?异常,也认为执行完毕,进行任务数量统计
w.completedTasks++;
w.unlock();
}
}
/**
* 执行到这里代表while循环结束了。worker线程正常结束了
* 1.workQueue中没有任务了(poll超时，或者使用take时通过shutdown、shutdownNow中断了)
* 2.worker线程执行时没有task出现异常，否则也会跳出循环
*/
//(10.7)执行到这里代表非核心线程在keepAliveTime内无法获取任务而退出
completedAbruptly = false;
} finally {
//(11)从上面可以看出如果实际业务(task任务)出现异常会导致当前worker终止
//completedAbruptly此时为true代表worker突然完成，不是正常退出
processWorkerExit(w, completedAbruptly);
}
}
其中processWorkerExit方法的源码如下：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) //如果是task异常导致结束，workerCount数需要减1
decrementWorkerCount();
//(11.1)统计整个线程池完成的任务个数，并从works集合中删除当前worker
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
//(11.2)尝试设置线程状态为TERMINATED
//条件一：如果当前是SHUTDWON状态并且工作队列为空
//或者
//条件二：当前状态是STOP状态并且当前线程池里面没有活动线程
tryTerminate();
//(11.3)如果当前线程个数小于核心线程数，则新增worker线程
int c = ctl.get();
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);
}
}
5、线程池获取任务
worker线程会通过自旋方式一直循环获取task任务，先执行自己携带的任务，如果自己携带的任务为空则从阻塞队列中获取任务。
//(10)自旋。先执行自己携带的任务，然后从阻塞队列中获取一个任务直到无法获取任务
while (task != null || (task = getTask()) != null){......}
其中，从阻塞队列中获取任务的getTask方法源码如下：
private Runnable getTask() {
boolean timedOut = false;//最近一次poll()超时了吗？
//(10.0.1)自旋获取任务(因为是多线程环境)
for (; ; ) {
int c = ctl.get();
int rs = runStateOf(c);
/**
* 1.线程池状态是SHUTDOWN状态并且任务队列是空
* 2.线程池是STOP以上的状态
* (10.0.2)满足以上两个条件则workerCount数-1，并且返回null从而保证获取任务的worker进行正常退出
*/
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
/**
* 1.允许核心线程退出
* 2.当前线程数量超过corePoolSize核心线程数
* (10.0.3)这时获取任务的机制切换为poll(keepAliveTime)
*/
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/**
* 1、线程数大于maximumPoolSize(什么时候会出现这种情况？ 当maximumPoolSize初始设置为0或者其他线程通过set方法对其进行修改)
* 2、线程数未超过maximumPoolSize但是timed为true(允许核心线程退出或者线程数量超过核心线程)并且上次获取任务超时(没获取到任务,我们推测本次依旧会超时)
* 3、在满足条件1或者条件2的情况下进行check：运行线程数大于1或者任务队列没有任务
* (10.0.4)满足以上条件执行worker线程数减1操作，并且返回null从而保证获取任务的worker进行正常退出
*/
if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//(10.0.5)如果允许超时退出，则调用poll(keepAliveTime)获取任务，否则则通过tack()一直阻塞等待直到有任务提交到队列
Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
if (r != null)
return r;
//(10.0.6)当等待超过keepAliveTime时间未获取到任务时，标记为true。在下次自旋时会进入销毁流程
timedOut = true;
} catch (InterruptedException retry) {
//(10.0.7)什么时候会抛出异常？当调用shutdown或者shutdownNow方法触发worker内的Thread调用interrupt方法时会执行到此处
timedOut = false;
}
}
}
6、线程池关闭操作
线程池提供了两种关闭线程池的方法：
- shutdown()：调用后，不可以再 submit 新的 task，已经 submit 的将继续执行。
- shutdwonNow()：调用后，试图停止当前正在执行的 task，并返回尚未执行的task的列表。6.1、 调用shutdown方法 public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock();//利用排它锁进行上锁，保证只有一个线程执行关闭流程 try { //(12)权限检查 checkShutdownAccess(); //(13)内部通过自旋+CAS修改线程池状态为SHUTDOWN advanceRunState(SHUTDOWN); //(14)遍历所有的worker，进行中断通知 interruptIdleWorkers(); onShutdown();//关闭线程池时调用的钩子函数 } finally { mainLock.unlock(); } //(15)进行最后的整理工作，尝试状态变为TERMINATED tryTerminate(); } //如果当前状态>=SHUTDOWN则直接返回，否则设置当前状态为SHUTDOWN private void advanceRunState(int targetState) { for (; ; ) { int c = ctl.get(); if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) break; } } //设置所有空闲线程的中断标志 private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread; //如果工作线程没有被中断，并且没有正在运行则设置中断 if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) break; } } finally { mainLock.unlock(); } } //尝试终止线程池 final void tryTerminate() { for (; ; ) { int c = ctl.get(); /** * 1.处于RUNNING状态 * 2.处于TIDYING、TERMINATED状态（已经终止过） * 3.处于SHUTDOWN状态但是workQueue不为空，还有任务未处理 * (15.1)满足以上任何1种条件线程池不能被终止 */ if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && !workQueue.isEmpty())) return;//当前线程池不能被为终止 /** * 1.处于SHUTDOWN状态并且workQueue为空，或者STOP状态 * 2.如果此时线程池还有线程(正在执行任务，正在等待任务)，中断1个空闲的worker线程 * (15.2)如果还有worker线程，只中断1个线程并返回 */ if (workerCountOf(c) != 0) { //有资格终止 interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //(15.3)当前已经没有运行态的线程了，将线程池状态设置为TIDYING，workerCount=0 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { //(15.4)钩子方法，待子类实现 terminated(); } finally { //(15.5)将线程池状态设置为TERMINATED，workerCount=0 ctl.set(ctlOf(TERMINATED, 0)); //(15.5)将线程池状态设置为TERMINATED后唤醒awaitTermination操作阻塞的线程 termination.signalAll(); } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } } 代码（15）判断如果当前线程池状态是SHUTDOWN状态并且工作队列为空或者当前是STOP状态当前线程池里面没有活动线程则设置线程池状态为TERMINATED，如果设置为了TERMINATED状态还需要调用条件变量termination的signalAll()方法激活所有因为调用线程池的awaitTermination()方法而被阻塞的线程。6.2、 调用shutdownNow方法调用shutdownNow()后，线程池就不会再接受新的任务，并且丢弃工作队列里面里面的任务，正在执行的任务会被中断，该方法是立刻返回的，并不等待激活的任务执行完成再返回。返回值为这时候队列里面被丢弃的任务列表。代码如下： public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //(16)权限检查 checkShutdownAccess(); //(17)设置线程池状态为STOP advanceRunState(STOP); //(18)中断所有线程 interruptWorkers(); //(19)移动任务队列到tasks中 tasks = drainQueue(); } finally { mainLock.unlock(); } tryTerminate(); return tasks; } //中断所有的worker线程，包含空闲线程和正在执行任务的线程 private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) w.interruptIfStarted(); } finally { mainLock.unlock(); } } 7、awaitTermination操作线程池调用awaitTermination(long timeout, TimeUnit unit)方法后，当前线程会被阻塞，直到线程池状态变为了TERMINATED才返回，或者等待时间超时才返回，整个过程独占锁,代码如下：
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (; ; ) {
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
8、线程池容量的动态调整
ThreadPoolExecutor提供了动态调整线程池容量大小的方法：
- setCorePoolSize(int corePoolSize)：设置核心池大小。 public void setCorePoolSize(int corePoolSize) { if (corePoolSize < 0) throw new IllegalArgumentException(); int delta = corePoolSize - this.corePoolSize; this.corePoolSize = corePoolSize; if (workerCountOf(ctl.get()) > corePoolSize) interruptIdleWorkers(); else if (delta > 0) { // We don't really know how many new threads are "needed". // As a heuristic, prestart enough new workers (up to new // core size) to handle the current number of tasks in // queue, but stop if queue becomes empty while doing so. int k = Math.min(delta, workQueue.size()); while (k-- > 0 && addWorker(null, true)) { if (workQueue.isEmpty()) break; } } }
- setMaximumPoolSize(int maximumPoolSize)：设置线程池最大容量。 public void setMaximumPoolSize(int maximumPoolSize) { if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize) throw new IllegalArgumentException(); this.maximumPoolSize = maximumPoolSize; if (workerCountOf(ctl.get()) > maximumPoolSize) interruptIdleWorkers(); } 当上述参数从小变大时，ThreadPoolExecutor进行线程赋值，还可能立即创建新的线程来执行任务。
五、线程池状态转换
首先，总结一下线程池的状态(这里的状态值取移位之前的）。
六、总结
如何选择线程池数量？ 影响线程池大小的因素： CPU的数量、内存大小、 任务计算密集型还是IO密集型等。 牛人总结的线程池计算公式如下：
- NCPU = CPU的数量
- UCPU = 期望对CPU的使用率 0 ≤ UCPU ≤ 1
- W/C = 等待时间与计算时间的比率如果希望处理器达到理想的使用率，那么线程池的最优大小为： 线程池大小=NCPU *UCPU(1+W/C) 一般需要根据任务的类型来配置线程池大小：
- 如果是CPU密集型任务，就需要尽量压榨CPU，参考值可以设为 NCPU+1
- 如果是IO密集型任务，参考值可以设置为2*NCPU
- 具体的设置还需要根据实际情况进行调整，比如可以先将线程池大小设置为参考值，再观察任务运行情况和系统负载、资源利用率来进行适当调整。
码字不易，且看且支持。原文地址：https://blog.csdn.net/u011047968/article/details/106937285
本文暂时没有评论，来添加一个吧(●'◡'●)