1. 线程池的类型
- newFixedThreadPool：该方法返回一个固定数量的线程池，线程数不变，当有一个任务提交时，若线程池中空闲，则立即执行，若没有，则会被暂缓在一个任务队列中，等待有空闲的线程去执行；
- newSingleThreadExecutor: 创建一个线程的线程池，若空闲则执行，若没有空闲线程则暂缓在任务队列中；
- newCachedThreadPool：返回一个可根据实际情况调整线程个数的线程池，不限制最大线程数量，若有空闲的线程则执行任务，若无任务则不创建线程。并且每一个空闲线程会在 60 秒后自动回收；
- newScheduledThreadPool: 创建一个可以指定线程的数量的线程池，但是这个线程池还带有延迟和周期性执行任务的功能，类似定时器。
2. ThreadPoolExecutor
上面提到的四种线程池的构建，都是基于 ThreadpoolExecutor 来构建的，接下来将说明ThreadPoolThread 有哪些构造参数
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}ThreadpoolExecutor 有多个重载的构造方法，我们可以基于它最完整的构造方法来分析先来解释一下每个参数的作用，稍后我们在分析源码的过程中再来详细了解参数的意义。
public ThreadPoolExecutor(int corePoolSize, //核心线程数量
int maximumPoolSize, //最大线程数
long keepAliveTime, //超时时间,超出核心线程数量以外的线程存活时间
TimeUnit unit, //存活时间单位
BlockingQueue<Runnable> workQueue, //保存执行任务的队列
ThreadFactory threadFactory,//创建新线程使用的工厂
RejectedExecutionHandler handler//当任务无法执行的时候的处理方式)线程池初始化时是没有创建线程的，线程池里的线程的初始化与其他线程一样，但是在完成任务以后，该线程不会自行销毁，而是以挂起的状态返回到线程池。直到应用程序再次向线程池发出请求时，线程池里挂起的线程就会再度激活执行任务。这样既节省了建立线程所造成的性能损耗，也可以让多个任务反复重用同一线程，从而在应用程序生存期内节约大量开销。
newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}FixedThreadPool 的核心线程数和最大线程数都是指定值，也就是说当线程池中的线程数超过核心线程数后，任务都会被放到阻塞队列中。另外 keepAliveTime 为 0，也就是超出核心线程数量以外的线程空余存活时间而这里选用的阻塞队列是 LinkedBlockingQueue，使用的是默认容量 Integer.MAX_VALUE，相当于没有上限。
这个线程池执行任务的流程如下：
- 线程数少于核心线程数，也就是设置的线程数时，新建线程执行任务；
- 线程数等于核心线程数后，将任务加入阻塞队列；
- 由于队列容量非常大，可以一直添加；
- 执行完任务的线程反复去队列中取任务执行。
用途：FixedThreadPool 用于负载比较大的服务器，为了资源的合理利用，需要限制当前线程数量
newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}CachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程; 并且没有核心线程，非核心线程数无上限，但是每个空闲的时间只有 60 秒，超过后就会被回收。
它的执行流程如下：
- 没有核心线程，直接向 SynchronousQueue 中提交任务；
- 如果有空闲线程，就去取出任务执行；如果没有空闲线程，就新建一个；
- 执行完任务的线程有 60 秒生存时间，如果在这个时间内可以接到新任务，就可以继续活下去，否则就被回收；
newSingleThreadExecutor
创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
3. 线程池的实现原理
线程池的基本使用我们都清楚了，接下来我们来了解一下线程池的实现原理 ThreadPoolExecutor 是线程池的核心，提供了线程池的实现。
ScheduledThreadPoolExecutor 继承了 ThreadPoolExecutor，并另外提供一些调度方法以支持定时和周期任务。Executors 是工具类，主要用来创建线程池对象我们把一个任务提交给线程池去处理的时候，线程池的处理过程是什么样的呢？首先直接来看看定义
线程池原理分析(FixedThreadPool)
4. 源码分析
execute
基于源码入口进行分析，先看 execute 方法
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {//1.当前池中线程比核心数少，新建一个线程执行任务
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {//2.核心池已满，但任务队列未满，添加到队列中
int recheck = ctl.get();
//任务成功添加到队列以后，再次检查是否需要添加新的线程，因为已存在的线程可能被销毁了
if (!isRunning(recheck) && remove(command))
reject(command);//如果线程池处于非运行状态，并且把当前的任务从任务队列中移除成功，则拒绝该任务
else if (workerCountOf(recheck) == 0)//如果之前的线程已被销毁完，新建一个线程
addWorker(null, false);
}
else if (!addWorker(command, false)) //3.核心池已满，队列已满，试着创建一个新线程
reject(command); //如果创建新线程失败了，说明线程池被关闭或者线程池完全满了，拒绝任务
}ctl 的作用
在线程池中，ctl 贯穿在线程池的整个生命周期中
ctl： private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));它是一个原子类，主要作用是用来保存线程数量和线程池的状态。我们来分析一下这段代码，其实比较有意思，他用到了位运算。一个 int 数值是 32 个 bit 位，这里采用高 3 位来保存运行状态，低 29 位来保存线程数量。
我们来分析默认情况下，也就是 ctlOf(RUNNING)运行状态，调用了 ctlOf(int rs,int wc)方法；
其中
private static int ctlOf(int rs, int wc) { return rs | wc; }其中 RUNNING =-1 << COUNT_BITS ； -1 左移 29 位，-1 的二进制是 32 个 1（1111 1111 1111
1111 1111 1111 1111 1111）
-1 的二进制计算方法
原码是 1000…001 . 高位 1 表示符号位。
然后对原码取反，高位不变得到 1111…110
然后对反码进行+1 ，也就是补码操作， 最后得到 1111…1111
那么-1 <<左移 29 位， 也就是 【111】 表示； rs | wc 。二进制的 111 | 000 。得到的结果仍然是 111
那么同理可得其他的状态的 bit 位表示
private static final int COUNT_BITS = Integer.SIZE - 3; //32-3
private static final int CAPACITY = (1 << COUNT_BITS) - 1; //将 1 的二进制向左位移 29 位,再减 1 表示最大线程容量
// 运行状态保存在 int 值的高 3 位 ( 所有数值左移 29 位 )
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; // 所有的任务都已结束,线程数量为 0,处于该状态的线程池即将调用 terminated()方法
private static final int TERMINATED = 3 << COUNT_BITS;// terminated()方法执行完成状态转化
addWorker
如果工作线程数小于核心线程数的话，会调用 addWorker，顾名思义，其实就是要创建一个工作线程。我们来看看源码的实现，源码比较长，看起来比较唬人，其实就做了两件事。
- 才用循环 CAS 操作来将线程数加 1；
- 新建一个线程并启用。
private boolean addWorker(Runnable firstTask, boolean core) {
retry: for (;;) { //goto 语句,避免死循环
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
//如果线程处于非运行状态，并且 rs 不等于 SHUTDOWN 且 firstTask 不等于空且且
//workQueue 为空，直接返回 false （表示不可添加 work 状态）
//1. 线程池已经 shutdown 后，还要添加新的任务，拒绝
//2. （第二个判断） SHUTDOWN 状态不接受新任务，但仍然会执行已经加入任务队列的任务，所以当进入 SHUTDOWN 状态，而传进来的任务为空，并且任务队列不为空的时候，是允许添加新线程的, 如果把这个条件取反，就表示不允许添加 worker
if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null &&
!workQueue.isEmpty()))
return false;
for (;;) { //自旋
int wc = workerCountOf(c);//获得 Worker 工作线程数
//如果工作线程数大于默认容量大小或者大于核心线程数大小，则直接返回 false 表示不能再添加 worker。
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))//通过 cas 来增加工作线程数，如果 cas 失败，则直接重试
break retry;
c = ctl.get(); // Re-read ctl // 再次获取 ctl 的值
if (runStateOf(c) != rs) //这里如果不想等，说明线程的状态发生了变化，继续重试
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//上面这段代码主要是对 worker 数量做原子+1 操作,下面的逻辑才是正式构建一个 worker
boolean workerStarted = false; //工作线程是否启动的标识
boolean workerAdded = false; //工作线程是否已经添加成功的标识
Worker w = null;
try {
w = new Worker(firstTask); //构建一个 Worker，这个 worker 是什么呢？我们可以看到构造方法里面传入了一个 Runnable 对象
final Thread t = w.thread; //从 worker 对象中取出线程
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());
//只有当前线程池是正在运行状态，[或是 SHUTDOWN 且 firstTask 为空]，才能添加到 workers 集合中
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//任务刚封装到 work 里面，还没 start,你封装的线程就是 alive，几个意思？肯定是要抛异常出去的
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w); //将新创建的 Worker 添加到 workers 集合中
int s = workers.size();
//如果集合中的工作线程数大于最大线程数，这个最大线程数表示线程池曾经出现过的最大线程数
if (s > largestPoolSize)
largestPoolSize = s; //更新线程池出现过的最大线程数
workerAdded = true;//表示工作线程创建成功了
}
} finally {
mainLock.unlock(); //释放锁
}
if (workerAdded) {//如果 worker 添加成功
t.start();//启动线程
workerStarted = true;
}
}
} finally {
if (!workerStarted)
addWorkerFailed(w); //如果添加失败，就需要做一件事，就是递减实际工作线程数(还记得我们最开始的时候增加了工作线程数吗)
}
return workerStarted;//返回结果
}Worker 类说明
我们发现 addWorker 方法只是构造了一个 Worker，并且把 firstTask 封装到 worker 中，它是做什么的呢？我们来看看
- 每个 worker,都是一条线程，同时里面包含了一个 firstTask，即初始化时要被首先执行的任务；
- 最终执行任务的,是 runWorker()方法。
Worker 类继承了 AQS，并实现了 Runnable 接口，注意其中的 firstTask 和 thread 属性：
- firstTask 用它来保存传入的任务；
- thread 是在调用构造方法时通过 ThreadFactory 来创建的线程，是用来处理任务的线程。
在调用构造方法时，需要传入任务，这里通过 getThreadFactory().newThread(this)；来新建一个线程，newThread 方法传入的参数是 this，因为 Worker 本身继承了 Runnable 接口，也就是一个线程，所以一个 Worker 对象在启动的时候会调用 Worker 类中的 run 方法。
Worker 继承了 AQS，使用 AQS 来实现独占锁的功能。为什么不使用 ReentrantLock 来实现呢？可以看到 tryAcquire 方法，它是不允许重入的，而 ReentrantLock 是允许重入的；lock 方法一旦获取了独占锁，表示当前线程正在执行任务中；那么它会有以下几个作用：
- 如果正在执行任务，则不应该中断线程；
- 如果该线程现在不是独占锁的状态，也就是空闲的状态，说明它没有在处理任务，这时可以对该线程进行中断；
- 线程池在执行 shutdown 方法或 tryTerminate 方法时会调用 interruptIdleWorkers 方法来中断空闲的线程，interruptIdleWorkers 方法会使用 tryLock 方法来判断线程池中的线程是否是空闲状态；
- 之所以设置为不可重入，是因为我们不希望任务在调用像 setCorePoolSize 这样的线程池控制方法时重新获取锁，这样会中断正在运行的线程。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread; //注意了，这才是真正执行 task 的线程，从构造函数可知是由 ThreadFactury 创建的
/** Initial task to run. Possibly null. */
Runnable firstTask; //这就是需要执行的 task
/** Per-thread task counter */
volatile long completedTasks; //完成的任务数，用于线程池统计
Worker(Runnable firstTask) {
setState(-1); // 初始状态 -1, 防止在调用 runWorker()，也就是真正执行 task前中断thread
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
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) {
}
}
}
}addWorkerFailed
addWorker 方法中，如果添加 Worker 并且启动线程失败，则会做失败后的处理。这个方法主要做三件事：
- 如果 worker 已经构造好了，则从 workers 集合中移除这个 worker；
- 原子递减核心线程数（因为在 addWorker 方法中先做了原子增加）；
- 尝试结束线程池。
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}runWorker方法
前面已经了解了 ThreadPoolExecutor 的核心方法 addWorker，主要作用是增加工作线程，而 Worker 简单理解其实就是一个线程，里面重新了 run 方法，这块是线程池中执行任务的真正处理逻辑，也就是 runWorker 方法，这个方法主要做几件事：
- 如果 task 不为空，则开始执行 task；
- 如果 task 为空，则通过 getTask() 再去取任务，并赋值给 task，如果取到的 Runnable 不为空，则执行该任务；
- 执行完毕后,通过 while 循环继续 getTask() 取任务；
- 如果 getTask() 取到的任务依然是空，那么整个 runWorker()方法执行完毕。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
//unlock，表示当前 worker 线程允许中断，因为 new Worker 默认的 state=-1,此处是调用Worker 类的 tryRelease()方法，将 state 置为 0，而 interruptIfStarted()中只有 state>=0 才允许调用中断
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//注意这个 while 循环,在这里实现了 [线程复用] // 如果 task 为空，则通过getTask 来获取任务
while (task != null || (task = getTask()) != null) {
w.lock(); //上锁，不是为了防止并发执行任务，为了在 shutdown()时不终止正在运行的 worker线程池为 stop 状态时不接受新任务，不执行已经加入任务队列的任务，还中断正在执行的任务
// 所以对于 stop 状态以上是要中断线程的
//(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP) 确保线程中断标志位为 true 且是 stop 状态以上，接着清除了中断标志
//!wt.isInterrupted() 则再一次检查保证线程需要设置中断标志位
if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);//这里默认是没有实现的，在一些特定的场景中我们可以自己继承 ThreadpoolExecutor 自己重写
Throwable thrown = null;
try {
task.run(); //执行任务中的 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,需要再通过 getTask()取) + 记录该 Worker 完成任务数量 + 解锁
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
//1.将入参 worker 从数组 workers 里删除掉；
//2.根据布尔值 allowCoreThreadTimeOut 来决定是否补充新的 Worker 进数组
workers
}
}getTask
worker 线程会从阻塞队列中获取需要执行的任务，这个方法不是简单的 take 数据，我们来分析下他的源码实现，你也许好奇是怎样判断线程有多久没有活动了，是不是以为线程池会启动一个监控线程，专门监控哪个线程正在偷懒？想太多，其实只是在线程从工作队列 poll 任务时，加上了超时限制，如果线程在 keepAliveTime 的时间内 poll 不到任务，那我就认为这条线程没事做，可以干掉了，看看这个代码片段你就清楚了
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {//自旋
int c = ctl.get();
int rs = runStateOf(c);
/**对线程池状态的判断，两种情况会 workerCount-1，并且返回 null
*1. 线程池状态为 shutdown，且 workQueue 为空（反映了 shutdown 状态的线程池还是要执行 workQueue 中剩余的任务的）
*2. 线程池状态为 stop（shutdownNow()会导致变成 STOP）（此时不用考虑 workQueue的情况）
*/
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;//返回 null，则当前 worker 线程会退出
}
int wc = workerCountOf(c);
// timed 变量用于判断是否需要进行超时控制。
// allowCoreThreadTimeOut 默认是 false，也就是核心线程不允许进行超时；
// wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量；
// 对于超过核心线程数量的这些线程，需要进行超时控制
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//1. 线程数量超过 maximumPoolSize 可能是线程池在运行时被调用了 setMaximumPoolSize()被改变了大小，否则已经 addWorker()成功不会超过 maximumPoolSize
//2. timed && timedOut 如果为 true，表示当前操作需要进行超时控制，并且上次从阻塞队列中获取任务发生了超时.其实就是体现了空闲线程的存活时间
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//根据 timed 来判断，如果为 true，则通过阻塞队列 poll 方法进行超时控制，如果在keepaliveTime 时间内没有获取到任务，则返回 null. 否则通过 take 方法阻塞式获取队列中的任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
if (r != null)//如果拿到的任务不为空，则直接返回给 worker 进行处理
return r;
timedOut = true;//如果 r==null，说明已经超时了，设置 timedOut=true，在下次自旋的时候进行回收
} catch (InterruptedException retry) {
timedOut = false; // 如果获取任务时当前线程发生了中断，则设置 timedOut为false并返回循环重试
}
}
}这里重要的地方是第二个 if 判断，目的是控制线程池的有效线程数量。由上文中的分析可以知道，在执行 execute 方法时，如果当前线程池的线程数量超过了 corePoolSize 且小于maximumPoolSize，并且 workQueue 已满时，则可以增加工作线程，但这时如果超时没有获取到任务，也就是 timedOut 为 true 的情况，说明 workQueue 已经为空了，也就说明了当前线程池中不需要那么多线程来执行任务了，可以把多于 corePoolSize 数量的线程销毁掉，保持线程数量在 corePoolSize 即可。
什么时候会销毁？当然是 runWorker 方法执行完之后，也就是 Worker 中的 run 方法执行完，由 JVM 自动回收。
getTask 方法返回 null 时，在 runWorker 方法中会跳出 while 循环，然后会执行 processWorkerExit 方法。
processWorkerExit
runWorker 的 while 循环执行完毕以后，在 finally 中会调用 processWorkerExit， 来销毁工作线程。
到目前为止，我们已经从 execute 方法中输入了 worker 线程的创建到执行以及最后到销毁的全部过程。那么我们继续回到 execute 方法。我们只分析完 addWorker 这段逻辑，继续来看后面的判断
execute 后续逻辑分析
如果核心线程数已满，说明这个时候不能再创建核心线程了，于是走第二个判断
第二个判断逻辑比较简单，如果线程池处于运行状态并且任务队列没有满，则将任务添加到队列中；
第三个判断，核心线程数满了，队列也满了，那么这个时候创建新的线程也就是（非核心线程），如果非核心线程数也达到了最大线程数大小，则直接拒绝任务。
if (isRunning(c) && workQueue.offer(command)) {//2.核心池已满，但任务队列未满，添加到队列中
int recheck = ctl.get();
//任务成功添加到队列以后，再次检查是否需要添加新的线程，因为已存在的线程可能被销毁了
if (! isRunning(recheck) && remove(command))
reject(command);//如果线程池处于非运行状态，并且把当前的任务从任务队列中移除成功，则拒绝该任务
else if (workerCountOf(recheck) == 0)//如果之前的线程已被销毁完，新建一个线程
addWorker(null, false);
}
else if (!addWorker(command, false)) //3.核心池已满，队列已满，试着创建一个新线程
reject(command); //如果创建新线程失败了，说明线程池被关闭或者线程池完全满了，拒绝任务5. 拒绝策略
- AbortPolicy：直接抛出异常，默认策略；
- CallerRunsPolicy：用调用者所在的线程来执行任务；
- DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
- DiscardPolicy：直接丢弃任务；
当然也可以根据应用场景实现 RejectedExecutionHandler 接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。
