目录
- 1 带着问题去阅读
- 1.1 线程池的线程复用原理
- 1.2 线程池如何管理线程
- 1.3 线程池配置的重要参数
- 1.4 shutdown()和shutdownNow()区别
- 1.5 线程池中的两个锁
- 2 源码分析过程中的困惑及解惑
- 3 源码分析
- 3.1 类继承关系
- 3.2 类的常量/成员变量
- 3.3 成员变量访问方法
- 3.4 构造函数
- 3.5 静态内部类Worker
- 3.5.1 Worker继承关系
- 3.5.2 Worker源码分析
- 4 重要方法详解
- 4.1 execute()方法
- 4.2 addWorker()方法
- 4.3 runWorker()方法
- 4.4 getTask()方法
- 4.5 processWorkerExit()方法
- 4.6 tryTerminate()方法
- 4.7 interruptIdleWorker()方法
- 4.8 shutdown()方法
- 4.9 shutdownNow()方法
- 4.10 isShutdown()方法
- 4.11 prestartCoreThread()方法
- 4.12 prestartAllCoreThreads()方法
1 带着问题去阅读1.1 线程池的线程复用原理
用户每次调用execute()来提交一个任务，然后任务包装成Worker对象，并且启动一个worker线程来执行任务（任务可能会被先加入队列），只要任务队列不为空且worker线程没有被中断，线程的run()方法通过一个while循环，不断去队列获取任务并执行，而不会进入到run()方法底部。while循环是线程复用的关键
1.2 线程池如何管理线程
首先定义两个说明：
关于获取任务超时，会依赖以下条件:
–1、开启核心线程超时设置 或 线程池线程数大于核心线程数
–2、符合1，且从workqueue获取任务超时。（如果不符合1，则以阻塞方式获取任务，不会超时）线程池最小保留线程数：
–1、如果没有开启核心线程超时配置，则至少保留corePoolSize个线程
–2、如果开启核心线程超时并且当前队列里面还有任务，只需保留1个线程
将线程池的生命周期分为三个阶段：创建阶段、运行期间、终止阶段。
一、创建阶段
- 当线程池线程数（ctl低位）少于核心线程数（corePoolSize），创建新线程执行任务
- 当线程池线程数大于等于核心线程数，且任务队列未满时，将新任务放入到任务队列中，不创建线程
- 当线程池线程数大于等于核心线程数（maximumPoolSize），且任务队列已满
–如果工作线程数少于最大线程数，则创建新线程执行任务
–如果工作线程数等于最大线程数，则抛出异常，拒绝新任务进入
二、运行期间
1、线程启动后，将一直循环获取任务并执行，只有当获取任务超时，或者线程池被终止，才会结束。
2、如果获取任务超时，那么Worker线程自然结束。此时线程池减少了1个线程。
3、在线程结束后，线程池会检查：1、线程池线程数<最少保留线程数 2、任务执行异常结束。如果符合，线程池会自动补充1个Worker
三、终止阶段
调用shutdown()和shutdownNow()都导致线程池线程数减少。
1、shutdown()方式终止线程池：
–停止提交新的任务，已在队列的任务会继续执行，并且中断空闲的Worker线程(Work.state从0->1成功)，线程池状态变为SHUTDOWN
2、shutdownNow()方式终止线程池：
–关闭线程池，不再接受新的任务，中断已经启动的Worker线程（Work.state>0），线程池状态改为STOP
线程池创建线程及处理任务过程：
梳理一下大概流程：
- 用户线程调用execute()提交Runnable任务
- execute()调用addWork()将任务提交给线程池处理：如果有可用的核心线程，则提交给核心线程处理。反则，将任务先添加到任务队列（workQueen）中。
- addWorker()方法将启动一个worker线程，调用runWorker()来处理任务。
- runWorker()方法将循环获取任务，并运行任务的run()方法来执行真正的业务。如果是以核心线程提交任务，则优先处理该任务，否则，循环调用getTask()来获取任务
- getTask()方法，从任务队列（workQueen）取出任务，并返回。
- getTask()没有拿到任务，则执行线程结束processWorkerExit()
线程池创建阶段：
1.3 线程池配置的重要参数
- ctl：存储线程池状态以及线程数
- corePoolSize、maximumPoolSize、keepAliveTime、workQueue 参照下面的源码分析说明
- allowCoreThreadTimeOut：是否开启核心线程超时。默认false，不在构造函数设置，需要调用方法设置
- HashSet workers：线程池终止时会从该集合找线程来中断，源码分析有说明
1.4 shutdown()和shutdownNow()区别
- shutdown() ：关闭线程池，不再接受新的任务，已提交执行的任务继续执行；中断所有空闲线程；将线程池状态改为SHUTDOWN
- ShutDownNow()：关闭线程池，不再接受新的任务，中断已经启动的Worker线程；将线程池状态改为STOP；返回未完成的任务队列
1.5 线程池中的两个锁
- mainLock主锁是可重入的锁，用来同步更新的成员变量
- Worker内部实现了一个锁，它是不可重入的，在shutdown()场景中，通过tryLock确保不会中断还没有开始执行或者还在执行中的worker线程。
2 源码分析过程中的困惑及解惑
—什么情况任务会提交失败？
同时符合以下条件，任务才会被提交：
- 线程池状态等于RUNNING状态；
- 如果任务队列已经满了，并且线程池线程数 少于 配置的线程池最大线程数（maximumPoolSize） 且小于线程池的最大支持线程数(CAPACITY)时。（如果队列没满，任务将会先加入到队列中）
特别说明：特殊情况会创建任务为空的Worker线程来帮助队列中的任务跑完
—核心线程数的意义？从测试结果看，他决定了工作线程最大并发数，但未代码验证
- 核心线程数决定提交任务什么时候会被放入到队列中：即线程池线程数>=核心线程数时。
- 核心线程数大小跟并发执行线程（任务）无关。也就是，它不决定工作线程最大并发数
- 核心线程数可以动态修改。（如果增大了，可能会马上创建新的Worker线程）
—线程池状态不是RUNNING,或者往workQueue添加worker失败，这是为什么还要提交任务
以下情况会创建任务为空的Worker线程来执行队列中的任务
- 当前线程池状态为shutdown，但是任务队列不为空，这时创建Worker线程来帮助执行队列的任务
- 当前线程池状态为running， 任务添加到队列后，接着线程池被关闭，并且从队列移除该任务失败，并且线程池线程数为0，这时创建Worker线程来确保刚提交的任务有机会执行。
—为什么runWorker()方法在执行任务前后加锁，但是线程依然能够并发？
- worker线程是通过创建Worker对象来创建的，在addWorke()的while循环创建了多个Worker对象，每个Worker对象都有自己的锁，Worker线程通过runWorker()访问的是当前对象的锁，因此Worker线程能够并发；
- 锁的意义是限制不能中断执行中的任务，因为主线程调用shutdown()和shutdownNow()方法时，会遍历WorkerSet的Worker对象，调用tryLock()，这时主线程和Worker线程竞争同一个锁。
3 源码分析3.1 类继承关系
- Executo接口：专门提交任务，只有一个execute()方法。Executor 提供了一种将任务的提交和任务的执行两个操作进行解耦的思路：客户端无需关注执行任务的线程是如何创建、运行和回收的，只需要将任务的执行逻辑包装为一个 Runnable 对象传递进来即可，由 Executor 的实现类自己来完成最复杂的执行逻辑
- ExecutorService接口：继承了Executor，扩展执行任务的能力。例如：获取任务的执行结果、取消任务等功能；提供了关闭线程池、停止线程池，以及阻塞等待线程池完全终止的方法,需要ThreadPoolExecutor实现
- AbstractExecutorServic类：实现了 ExecutorService ，是上层的抽象类，负责将任务的执行流程串联起来，从而使得下层的实现类 ThreadPoolExecutor只需要实现一个执行任务的方法即可
- ThreadPoolExecutor：可以看做是基于生产者-消费者模式的一种服务，内部维护的多个线程相当于消费者，提交的任务相当于产品，提交任务的外部就相当于生产者
3.2 类的常量/成员变量
//--------------------------常量部分------------------------ // 常量29。用在移位计算Integer.SIZE=32）private static final int COUNT_BITS = Integer.SIZE - 3; //29// 最大支持线程数 2^29-1：000 11111111111111111...private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 以下为线程池的四个状态，用32位中的前三位表示 // 011 terminated() 方法执行完成后,线程池的状态会转为TERMINATED.private static final int TERMINATED = 3 << COUNT_BITS;// 010 所有任务都销毁了,workCount=0的时候,线程池的装填在转换为TIDYING是,会执行钩子方法terminated()private static final int TIDYING = 2 << COUNT_BITS; //翻译为整理// 001 拒绝新的任务提交,清空在队列中的任务private static final int STOP = 1 << COUNT_BITS;// 000 拒绝新的任务提交,会将队列中的任务执行完,正在执行的任务继续执行.private static final int SHUTDOWN = 0 << COUNT_BITS;// 111 00000 00000000 00000000 00000000 线程运行中 【running状态值为负数最小】private static final int RUNNING = -1 << COUNT_BITS; //线程池的默认状态//------------------------变量部分------------------------// ctl存储线程池状态和线程池大小，那么用前3位表示线程池状态，后29位表示：线程池大小，即线程池线程数//线程池状态初始值为RUNNINGprivate final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//任务队列//保存不能马上执行的Runnable任务。//执行shutdownNow()时，会返回还在队列的任务private final BlockingQueue workQueue;// 主锁，对workers、largestPoolSize、completedTaskCount的访问都必须先获取该锁private final ReentrantLock mainLock = new ReentrantLock();// 包含池中的所有工作线程的集合。持有mainLock访问 // 创建Worker时，添加到集合// 线程结束时，从集合移除// 调用shutdown()时，从该集合中找到空闲线程并中断// 调用shutdownNow()时，从该集合中找到已启动的线程并中断private final HashSet workers = new HashSet();// 线程通信手段, 用于支持awaitTermination方法：等待所有任务完成,并支持设置超时时间,返回值代表是不是超时.private final Condition termination = mainLock.newCondition();// 记录workers历史以来的最大值。持有mainLock访问// 每次增加worker的时候,都会判断当前workers.size()是否大于最大值，大于则更新// 用于线程池监控的，作为重要指标private int largestPoolSize;// 计数所有已完成任务,持有mainLock访问// 每个worker都有一个自己的成员变量 completedTasks 来记录当前 worker 执行的任务次数, 当前线worker工作线程终止的时候, 才会将worker中的completedTasks的数量加入到 completedTaskCount 指标中.private long completedTaskCount;// 线程工厂private volatile ThreadFactory threadFactory;// 拒绝策略,默认四种AbortPolicy、CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy,建议自己实现,增加监控指标private volatile RejectedExecutionHandler handler;// keepAliveTime和allowCoreThreadTimeOut 是关于线程空闲是否会被销毁的配置// 关于空闲的说明：// 1、线程池在没有关闭之前，会一直向任务队列（workqueue）获取任务执行，如果任务队列是空的，在新任务提交上来之前，就会产生一个等待时间，期间，线程处于空闲状态// 2、向任务队列获取任务用：workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)，表示阻塞式获取元素，等待超时，则终止等待并返回false。通过判断poll()方法是true/falle来判定线程是否超时// 获取任务的等待时间 ，以下两种情况会使用到该值//1、如果启用allowCoreThreadTimeOut，那表示核心线程的空闲时间 // 2、当线程池内线程数超过corePoolSize，表示线程获取任务的等待时间private volatile long keepAliveTime;// 核心线程是否开启超时// false：表示核心线程一旦启动，会一直运行，直至关闭线程池。默认该值// true：表示核心线程处于空闲且时间超过keepAliveTime，核心线程结束后，将不再创建新线程// （默认的构造函数没有设置这个属性，需要手工调用allowCoreThreadTimeOut()方法来设置)private volatile boolean allowCoreThreadTimeOut; //核心线程数量//核心线程是指：线程会一直存活在线程池中，不会被主动销毁【如果核心线程开启超时，有可能被被销毁】。private volatile int corePoolSize;// 配置的线程池最大线程数private volatile int maximumPoolSize;// 默认拒绝策略 AbortPolicyprivate static final RejectedExecutionHandler defaultHandler = new AbortPolicy();// 安全控制访问（主要用于shutdown和 shutdownNow方法private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread");// 在threadPoolExecutor初始化的时候赋值,acc对象是指当前调用上下文的快照，其中包括当前线程继承的AccessControlContext和任何有限的特权范围，使得可以在稍后的某个时间点(可能在另一个线程中)检查此上下文。private final AccessControlContext acc;3.3 成员变量访问方法
// 获取当前线程池的状态(前3位)private static int runStateOf(int c) { return c & ~CAPACITY; }// 获取当前线程池中线程数(后29位)private static int workerCountOf(int c){ return c & CAPACITY; }// 更新状态和数量private static int ctlOf(int rs, int wc) { return rs | wc; }// 小于判断C是不是小于S,比如runStateLessThan(var,STOP),那var就只有可能是(RUNNING,SHUTDOWN)private static boolean runStateLessThan(int c, int s) { return c = Sprivate static boolean runStateAtLeast(int c, int s) { return c >= s;}// 判断状态是不是RUNNINGprivate static boolean isRunning(int c) { return c < SHUTDOWN;}关于-1<<29说明：
-1 << COUNT_BITS这里是-1往左移29位，稍微有点不一样，-1的话需要我们自己算出补码来-1的原码10000000 00000000 00000000 00000001-1的反码，负数的反码是将原码除符号位以外全部取反11111111 11111111 11111111 11111110-1的补码，负数的补码就是将反码+111111111 11111111 11111111 11111111关键了，往左移29位，所以高3位全是1就是RUNNING状态111 00000 00000000 00000000 000000003.4 构造函数
//corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue 这五个参数必须指定//最多参构造函数public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue 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;//默认使用SynchronousQueue this.keepAliveTime = unit.toNanos(keepAliveTime); //默认60S this.threadFactory = threadFactory; //默认使用DefaultThreadFactory this.handler = handler; }构造函数总结：
初始化：corePoolSize（核心线程池大小）、maximumPoolSize（线程池容纳最大线程数）、workQueue（任务队列）、threadFactory（线程工厂）、keepAliveTime（空闲线程存活时长）、handler（拒绝策略）AccessControlContext
3.5 静态内部类Worker3.5.1 Worker继承关系
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {}- –Worker继承于AbstractQueuedSynchronizer
Worker继承于AQS 为的就是自定义实现不可重入的特性（所以没有使用 synchronized 或者 ReentrantLock）来辅助判断线程是否处于执行任务的状态：在开始执行任务前进行加锁，在任务执行结束后解锁，以便在后续通过判断 Worker 是否处于锁定状态来得知其是否处于执行阶段
- — Worker实现Runnable接口
Worker实现Runnable接口，线程是通过getThreadFactory().newThread(this) 来创建的，即将 Worker 本身作为构造参数传给 Thread 进行初始化，所以在 thread 启动的时候 Worker 的 run() 方法就会被执行。
关于ThreadFactory说明：
public interface ThreadFactory { Thread newThread(Runnable r);}3.5.2 Worker源码分析
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { private static final long serialVersionUID = 6138294804551838833L; //线程类型的属性：thread，线程池启动工作线程，就是启动这个thread。 // 1、通过this.thread=getThreadFactory().newThread(this)，初始化了属性thread，this就是指Worker对象 //2、因为Worker类实现了Runnable接口，所以thread启动后，会运行Worker的run()方法，然后就去执行runWorker(this)方法 final Thread thread; //线程要执行的第1个任务（可能为 null) 它表示这个任务立即执行，不需要放到任务队列。在工作线程数= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } } }Worker类总结：
- 所谓的线程池，其实就是正在运行的多个Worker线程。
- Worker作为线程启动后，它实际执行的是通过execute()提交的Runnable任务（实际业务），worker线程通过一个while循环来不断获取并任务，从而达到线程复用的效果
- firstTask：线程要执行的第1个任务（可能为 null) 它表示这个任务立即执行，不需要放到任务队列。在 1、线程数<核心线程数 2、队列已满且线程池不在运行状态 这两个场景下。
4 重要方法详解4.1 execute()方法
execute()用来提交要运行的任务
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // 计算当前线程池的状态及线程数 // 1、线程池线程数小于配置的核心线程数 if (workerCountOf(c) < corePoolSize) { // 将任务提交给核心线程处理 if (addWorker(command, true)) return; //失败的情况：1、线程池已经被关闭、2、线程池线程数大于等于核心线程数 （不能以true的方式提交了 ） c = ctl.get(); // 重新获取线程池状态 } // 2、无空闲核心线程，将任务加入队列 // 再次确认线程池为RUNNING状态，将任务加入队列【非阻塞式，队列满了会立即返回false】 if (isRunning(c) && workQueue.offer(command)) { //任务加入队列成功 int recheck = ctl.get() ;//再次获取当前线程池状态(线程池可能被其它线程关闭了) //判断当前线程池状态是不是RUNNING状态,不是就从workQueue中删除command任务 if (! isRunning(recheck) && remove(command)) reject(command);//执行拒绝策略 //如果当前线程数是0（那证明还没有其他工作线程去处理这个任务）,那么刚刚的任务肯定在阻塞队列里面了,这 else if (workerCountOf(recheck) == 0) addWorker(null, false);//开启一个没有任务的Worker线程去执行队列的任务 } // 3 workQueue添加worker失败，即队列满了 //创建非核心线程并执行任务 else if (!addWorker(command, false)) //如果线程创建失败，说明要么是线程池当前状态!=RUNNING，或者是任务队列已满且线程总数达到最大线程数了 reject(command);//执行拒绝策略. }execute()总结
- 进行三次addWorker的尝试：
- addWorker(command, true)：创建任务并以核心线程执行
- 核心线程数达到上限， 创建任务添加到任务队列，不创建线程
- addWorker(null, false) ：任务添加到队列后，接着线程池被关闭，并且从队列移除该任务失败，并且线程池线程数为0，这时创建任务并以非核心线程执行
- addWorker(command, false) ：任务队列已满，创建非核心线程并执行
- 任务提交失败情况：线程池非RUNNING状态 并且 任务队列已满并且线程池线程数达到最大线程数（maximumPoolSize）
4.2 addWorker()方法
//TERMINATED >TIDYING > STOP > SHUTDOWN > RUNNING //创建新的线程执行当前任务 //firstTask: 指定新增线程执行的第一个任务或者不执行任务private boolean addWorker(Runnable firstTask, boolean core) { //外循环： retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 如果线程池状态是SHUTDOWN、STOP、TIDYING、TERMINATED就不允许提交。 // && 后面的特殊情况，线程池的状态是SHUTDOWN并且要要执行的任务为Null并且队列不是空，这种情况下是允许增加一个线程来帮助队列中的任务跑完的，因为shutdown状态下，允许执行完成阻塞队里中的任务 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && //execute()有addWorkder(null,false)的场景 ! workQueue.isEmpty())) return false; //内循环：cas修改工作线程数，同时判断能否添加work for (;;) { int wc = workerCountOf(c); //添加任务前，线程池线程数已达到上限，此时不允许添加。上限分这三种情况： // 1、最大支持线程数 // 2、以core=true提交时，配置的核心线程数。（返回false后，会以core=false再提交一次） // 3、以core=false提交时，配置的线程池可容纳最大线程数。 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) //使用core则上限为核心线程数，否则最大线程数 return false; //没超过上限，通过CAS的方式增加worker的数量（+1），增加成功就跳出外层循环 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); //获取最新的线程池状态，与刚开始的状态比较 // - 变了，就从外层循环重新执行，重新进行状态的检查。 // - 没变，从当前循环重新执行，重新执行CAS操作。 if (runStateOf(c) != rs) continue retry; } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //创建Worker，并给firstTask赋初值 w = new Worker(firstTask); final Thread t = w.thread; //拿到属性thread if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); //此处加锁：因为涉及属性：workers、largestPoolSize（可能） 更新 try { int rs = runStateOf(ctl.get()); //获取线程池最新状态 if (rs < SHUTDOWN || //如果当前状态是 largestPoolSize) largestPoolSize = s; workerAdded = true; //表示添加worker成功 } } finally { mainLock.unlock(); } if (workerAdded) { //启动worker线程。该线程会一直循环执行getTask()，直至返回null，线程才结束 t.start(); //执行runWorker() workerStarted = true; //表示线程已经跑起来了 } } } finally { if (! workerStarted) addWorkerFailed(w);//worker线程没成功启动，进入失败处理逻辑 } return workerStarted;//;返回当前worker是否启动成功。 }addWorker()总结：
- 检查线程池状态以确定能否提交任务
- 校验能否以核心线程的方式提交任务
- 线程池的状态是SHUTDOWN并且任务队列不是空，允许增加一个线程来帮助队列中的任务跑完，但不会提交任务
- 更新线程池线程数
- 超过线程池线程数峰值则更新峰值（largestPoolSize）
- 加锁（mainLock）来更新
- 启动worker线程
4.3 runWorker()方法
//执行任务final void runWorker(Worker w) { Thread wt = Thread.currentThread(); //runWorker()是由Worker.run()调用，因此wt就是worker线程 Runnable task = w.firstTask; //拿到firstTask并赋值给局部变量task w.firstTask = null; //firstTask置空 w.unlock(); // 将state设置为0。因为构造函数设成-1，在执行任务前置为0。 boolean completedAbruptly = true;//标识任务是不是立刻就完成了。 try { //循环：先执行firstTask（不为空），后续通过getTask()获取任务。 while (task != null || (task = getTask()) != null) { //任务执行前加锁，任务完成后解锁。 //任何地方可通过判断锁状态来确认worker是否执行中 w.lock(); //加锁。防止任务在执行过程中被中断。 //判断目的：确保线程池当状态值大于等于 STOP 时有向线程发起过中断请求【调用了shutdownNow()】 // 两种情况: //1)如果当前线程池的状态是>=Stop的，并且当前线程没有被中断，那么就要执行中断。 //2)或者当前线程目前是已中断的状态并且线程池的状态也是>=Stop的（注意Thread.interrupted是会擦除中断标识符的），那么因为中断标识符已经被擦除了，那么!wt.isInterrupted()一定返回true，这个时候还是要将当前线程中断。第二次执行runStateAtLeast(ctl.get(), STOP)相当于一个二次检查 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt();//中断worker线程 。因为线程池将要终止了，所以这里没有从workerSet移除当前线程 try { beforeExecute(wt, task);//前置操作，空方法，可以业务自己实现 Throwable thrown = null; try { //执行任务：就是执行通过execute()提交的Runnable task.run();//第一个是firstTask，后面的是通过getTask()拿到的任务 } 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;//最后将task置为null，触发while循环的条件getTask() w.completedTasks++; //已完成的任务计数器+1 w.unlock();//释放当前线程的独占锁 } } completedAbruptly = false; //当第一个try的代码块有异常， completedAbruptly = false 不生效。最后completedAbruptly为true表示发生未知异常了 } finally { //getTask返回null时，执行任务退出 processWorkerExit(w, completedAbruptly);//completedAbruptly=true表示是突然退出的 } }runWorker()总结：
- 执行任务前先判断线程池是否是STOPING状态，是则中断worker线程。
- 执行任务：先执行firstTask，再从任务队列获取执行
- 如果没有任务，调用processWorkerExit()来执行线程退出的工作。
- 只要还有任务，worker线程就一直执行任务，并刷新completedTasks
4.4 getTask()方法
private Runnable getTask() { boolean timedOut = false; for (;;) { int c = ctl.get(); int rs = runStateOf(c); //1、先判断能否获取到任务 // 1)如果线程池的状态是>=STOP状态,这个时候不再处理队列中的任务,并且减少worker记录数量,返回的任务为null,这个时候在runRWorker方法中会执行processWorkerExit进行worker的退出操作. // 2)如果线程池的状态是>=SHUTDOWN并且workQueue为空,就说明处于SHOTdown以上的状态下,且没有任务在等待,那么也属于获取不到任务,getTask返回null. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount();//扣减线程池线程数，在processWorkerExit()处理线程退出 return null; } int wc = workerCountOf(c);//获取当前wokrer的数量 //以下涉及空闲线程是否会被线程池销毁的处理逻辑 // 线程超时处理前置条件：开启核心线程超时 或 线程池线程数大于核心线程数 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //线程超时处理的进一步判断： // 线程池线程数超过maximumPoolSize 或者 线程设置允许超时且当前worker取任务超时 //并且 // 线程池大小不是零或阻塞队列是空的),这种就返回null,并减少线程池线程计数 // 1、 (wc>maximumPoolSize) && (wc>1) 一般情况，线程池线程数会少于配置的最大线程数，但在addWork中 状态=shutdown且队列不为空时，会创建一个Worker，此时可能导致wc>maximumPoolSize，这里同时限定wc>1。因此线程池减少1个线程也不影响任务的执行【processWorkerExit()会保证还有任务就至少留有1个worker线程】。 // 2、 (wc>maximumPoolSize) && (workQueue.isEmpty()) 没有任务了，扣减更不影响 // 3 、(timed && timedOut) && (wc > 1) 超时了，先扣减再说 // 4 、(timed && timedOut) && (workQueue.isEmpty()) 超时了&队列没有任务，必须要扣减 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { //这里为啥不用decrementWorkerCount()呢,上面使用decrementWorkerCount()是因为确定不管是什么情况下,数量都要减,多减一次也没事,因为这个时候就是要关闭线程池释放资源 //这里不一样,线程池的状态可能是RUNNING状态,多减一次,可能导致获取不到worker去跑 if (compareAndDecrementWorkerCount(c)) return null; //扣减线程池线程数，在processWorkerExit()处理线程退出 continue;//扣减失败， 跳出本次循环重新检查 } //从队列中获取任务 //符合【线程超时处理前置条件】时用poll设置超时时间，不符合就使用take（阻塞直至有返回） try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; //task不为空，此处返回task timedOut = true; // 此处，r == null，肯定是poll操作超时了（注意，不代表队列空了），继续for循环，回到if ((wc > maximumPoolSize || (timed && timedOut)) 这个地方退出循环 } catch (InterruptedException retry) { timedOut = false; } } } private void decrementWorkerCount() { do {} while (! compareAndDecrementWorkerCount(ctl.get())); }getTask()总结：
- 从workQueue中获取一个任务并返回
- 没有获取到任务就扣减线程池线程数。获取不到任务的四种情况：
- 线程池的状态是>=STOP
- 线程池的状态是SHUTDOWN并且任务队列为空
- 获取任务超时
- 线程池线程数大于maximumPoolSize并且队列为空
4.5 processWorkerExit()方法
//worker线程没有拿到任务，成为空闲线程。该方法对空闲线程进一步处理private void processWorkerExit(Worker w, boolean completedAbruptly) { //如果completedAbruptly为true，则说明线程执行时出现异常，需要将workerCount数量减一 //如果completedAbruptly为false，说明在getTask方法中已经对workerCount进行减一，这里不用再减 if (completedAbruptly) decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //更新已完成任务的数量的统计项 completedTaskCount += w.completedTasks; //从worker集合中移除该worker workers.remove(w); } finally { mainLock.unlock(); } //尝试关闭线程池,但如果是正常运行状态,就不会关闭 tryTerminate(); int c = ctl.get(); if (runStateLessThan(c, STOP)) {//1、线程池是SHUTDOWN或RUNNING（如果不是这两个状态,说明线程已经停止了，不做任何操作） if (!completedAbruptly) {//2、线程正常结束 // 如果没有开启核心线程超时配置，则至少保留corePoolSize个线程； int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && !workQueue.isEmpty())//如果允许核心线程超时并且当前队列里面还有任务没跑，必须留1个线程,不能全死掉. min = 1; // 如果线程池数量>=最少预留线程数 if (workerCountOf(c) >= min) return; // 线程自然结束了，不用补充worker } // 1、执行任务异常结束的，补充worker // 2、如果线程池数量<最少预留线程数，补充worker addWorker(null, false);//异常结束 增加worker //注: 别问我为啥上面要删除worker,还要再加,不删是不是不用加了. 明确下那个任务已经退出getTask那块的死循环了,永远回不去了,只能新增worker. }}processWorkerExit()方法总结！！！！！：
- 当Worker线程结束前，完成以下工作：扣减线程池线程数（ctl）、更新已完成任务数（completedTaskCount）、Worker集合中移除一个Worker（workers）、尝试终止线程池、计算线程池的最少保留线程数、根据最少保留线程数来确定是否补充一个Worker。
- 关于最少保留线程数：如果没有开启核心线程超时配置，则至少保留corePoolSize个线程；如果开启核心线程超时并且当前队列里面还有任务，只需保留1个线程；
- 需要补充worker的两种情况：1、线程池线程数<最少保留线程数 2、任务执行异常结束
4.6 tryTerminate()方法
//尝试终止线程池final void tryTerminate() { for (;;) { //cas自旋 确保更新成功 int c = ctl.get(); //RUNNING状态,不能终止线程池 //线程池状态是TIDYING或TERMINATED说明线程池已经处于正在终止的路上,不用再终止了. //状态为SHUTDOWN，但是任务队列不为空,也不能终止线程池 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; //调用shutdown()或者shutdownNow()方法时，执行以下处理 //工作线程数量不等于0，中断一个空闲的工作线程并返回 //这个时候线程池一定是 1、STOP的状态或者 2、SHUTDOW且队列为空 这两种情况中断一个空闲worker if (workerCountOf(c) != 0) { interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 设置线程池状态为TIDYING，如果设置成功，则调用terminated() if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { terminated(); //钩子方法，子类实现。默认什么都不做 } finally { // 设置状态为TERMINATED ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); //唤醒阻塞等待的线程 （future的场景） } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } }tryTerminate()总结
- 尝试终止线程池
- 不能终止线程池：
- 状态是RUNNING，不能直接终止（如果是调用shutdown()，shutdownNow()，会先将状态改为SHUTDOWN）
- 状态是TIDYING或者TERMINATED，不能终止（因为已经处于终止过程中）
- 状态是SHUTDOWN并且任务队列不为空，不能终止（因为还有任务要处理）
- 可以终止线程池：
- 状态是SHUTDOWN并且任务队列为空
- 状态是STOP
- 符合可以终止线程池的条件下，如果线程池线程数不等于0，那就中断1个Worker线程，不修改线程池状态
- 符合可以终止线程池的条件下，并且线程池线程数等于0，那就将线程池状态改为TIDYING，执行完钩子方法terminated()后状态再改为TERMINATED
interruptIdleWorkers(ONLY_ONE); 是否好奇为啥这里只中断一个worker呢, 这里就涉及到了线程池的优雅退出了.当执行到 interruptIdleWorkers(ONLY_ONE) 前面的时候, 线程池只能处于两种状态:1) STOP 状态 , 这个时候 workQueue 可能是有值的 , workQueue 在清空的过程中了.2) SHUTDOWN 状态并且 workQueue 是空的 .这两种状态都是说明, 线程池即将关闭, 或者说空闲的线程此时已经没用了,这个时候随手关一个, 反正要关,早关晚关而已.4.7 interruptIdleWorker()方法
//中断一个或多个线程private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //遍历worker，根据onlyOne判断，如果为ture只中断一个线程 for (Worker w : workers) { Thread t = w.thread; //线程没有被中断并且线程是空闲状态 //通过tryLock实现：不能中断还没有开始执行或者还在执行中的worker线程。 //线程未启动：-1 ，线程正在执行：1 ，trylock：0->1 ; if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); //中断操作，之后该线程就结束了 } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) break; } } finally { mainLock.unlock(); }}interruptIdleWorker()总结：
- 从worker集合中遍历并中断worker线程
- 只有worker线程状态是0的，才能够中断（不能中断未启动或者还在执行中的Worker线程）
4.8 shutdown()方法
//初始化一个有序的关闭，之前提交的任务都会被执行，但是新提交的任务则不会被允许放入任务队列中。如果之前被调用过了的话，那么再次调用也没什么用public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); //mainLock是全局变量，加锁确保不会并发关闭线程池 try { checkShutdownAccess();//安全策略判断。方法检查每一个线程池的线程是否有可以ShutDown的权限。 advanceRunState(SHUTDOWN); //CAS自旋把ctl中的状态从RUNNING变为SHUTDOWN interruptIdleWorkers();//中断所有空闲线程 onShutdown(); // 方法告知子类，线程池要处于ShutDown状态了 ，ScheduledThreadPoolExecutor预留的钩子 } finally { mainLock.unlock(); } tryTerminate();//尝试终止线程池 }shutdown()方法总结
- 执行shutdown()方法：关闭线程池，不再接受新的任务，已提交执行的任务继续执行。
- 调用interruptIdleWorkers()先中断所有空闲线程
- 调用tryTerminate()尝试终止线程池
- shutdown()将线程池状态改为SHUTDOWN但不是STOP
4.9 shutdownNow()方法
//关闭线程池，不再接受新的任务，正在执行的任务尝试终止public List shutdownNow() { List tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(STOP);//线程池的状态置为STOP interruptWorkers(); tasks = drainQueue(); //将剩余任务返回 } finally { mainLock.unlock(); } tryTerminate(); return tasks; } private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) //循环所有的worker w.interruptIfStarted();//已经启动的线程直接执行中断 } finally { mainLock.unlock(); } } void interruptIfStarted() { Thread t; //只有刚刚构建的worker的时候,状态state值是-1(这里也能体现刚构建的worker无法被中断),其他情况都是>=0的 if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } }ShutDownNow()方法总结
- 关闭线程池，不再接受新的任务，中断已经启动的Worker线程
- 将线程池状态改为STOP
- 返回未完成的任务队列
4.10 isShutdown()方法
确认线程池是否关闭。判断状态是不是RUNNING.
public boolean isShutdown() { return ! isRunning(ctl.get()); }4.11 prestartCoreThread()方法
public boolean prestartCoreThread() { return workerCountOf(ctl.get()) < corePoolSize && addWorker(null, true); }- 启动一个空闲的线程作为核心线程
- 如果核心线程数已到阈值, 会加入失败, 返回false, 如果线程池处于SHUTDOWN以上的状态也返回false
- 只有真正这个线程调用start方法跑起来, 才会返回true
4.12 prestartAllCoreThreads()方法
启动所有核心线程，使他们等待获取任务
public int prestartAllCoreThreads() { int n = 0; while (addWorker(null, true))//null代表空闲线程,true代表是增加的是核心线程 ++n;//死循环增加空闲 worker 而已 return n; }