线程
线程是调度CPU资源的最小单位，线程模型分为KLT模型与ULT模型，JVM使用的KLT模型，Java线程与OS线程保持 1:1的映射关系，也就是说一个Java线程也会在操作系统里有一个对应的线程。Java线程有多种生命状态：
image.png
为什么要使用线程池
用户态和内核态的概念
为了限制不同的程序之间的访问能力, 防止他们获取别的程序的内存数据, 或者获取外围设备的数据, 并发送到网络, CPU划分出两个权限等级 ：用户态和内核态。
用户态和内核态直接的关系：
image.png
用户态和内核态的切换
当在系统中执行一个程序时，大部分时间是运行在用户态下的，在其需要操作系统帮助完成一些用户态自己没有特权和能力完成的操作时就会切换到内核态。
线程状态切换的代价
Java的线程是映射到操作系统原生线程之上的，如果要阻塞或唤醒一个线程就需要操作系统介入，需要在用户态与核心态之间切换，这种切换会消耗大量的系统资源，因为用户态与内核态都有各自专用的内存空间，专用的寄存器等，用户态切换至内核态需要传递给许多变量、参数给内核，内核也需要保护好用户态在切换时的一些寄存器值、变量等，以便内核态调用结束后切换回用户态继续工作。
synchronized会导致争用不到锁的线程进入阻塞状态，所以说它是Java语言中一个重量级的同步操纵，被称为重量级锁，为了缓解上述性能问题，JVM从1.5开始，引入了轻量锁与偏向锁，默认启用了自旋锁，他们都属于乐观锁。所以明确java线程切换的代价，是理解java中各种锁的优缺点的基础之一。
线程池优势
- 重用存在的线程，减少线程创建，消亡的开销，提高性能。
- 提高响应速度。当任务到达时，从池中取出空闲线程立即执行，省去了等待线程创建的时间。
- 提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3; // 32 - 3
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 00011111 11111111 11111111 11111111
// 线程池处在RUNNING状态时，能够接收新任务，以及对已添加的任务进行处理。
private static final int RUNNING = -1 << COUNT_BITS; // 11100000 00000000 00000000 00000000
// 线程池处在SHUTDOWN状态时，不接收新任务，但能处理已添加的任务。
private static final int SHUTDOWN = 0 << COUNT_BITS; // 00000000 00000000 00000000 00000000
// 线程池处在STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。
private static final int STOP = 1 << COUNT_BITS; // 00100000 00000000 00000000 00000000
// 当所有的任务已终止，ctl记录的”任务数量”为0，线程池会变为TIDYING状态，并且执行钩子函数terminated()
// 若用户想在线程池变为TIDYING时，进行相应的处理，可以通过重载terminated()函数来实现。
private static final int TIDYING = 2 << COUNT_BITS; // 01000000 00000000 00000000 00000000
// 线程池彻底终止
private static final int TERMINATED = 3 << COUNT_BITS; // 01100000 00000000 00000000 00000000
image.png
而 ctl 的其余位所代表的数值则指定了当前线程池中正在执行任务的线程数（workerCount）。如下是操作 ctl 属性的相关方法：
// 用于获取当前线程池的状态，c为当前线程池工作时的ctl属性值
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 用于获取当前线程池正在工作的线程数量，c为当前线程池工作时的ctl属性值
private static int workerCountOf(int c) { return c & CAPACITY; }
// 这里rs表示当前线程的工作状态，wc则表示正在工作的线程数，该方法用于将这两个参数组装为一个ctl属性值
private static int ctlOf(int rs, int wc) { return rs | wc; }
// 当前线程池状态是否未达到指定状态，状态流转在数值上是依次增大的，因而这里只需要判断其大小即可
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 当前线程池状态是否至少处于某种状态
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
// 当前线程池是否处于正常运行状态
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
再来看一下它的构造方法：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
this.acc = System.getSecurityManager() == null ? null : AccessController.getContext();
/**
* 线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize
* 如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行
* 如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程
*/
this.corePoolSize = corePoolSize;
/**
* 线程池中允许的最大线程数。
* 如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize；
*/
this.maximumPoolSize = maximumPoolSize;
/**
* 用来保存等待被执行的任务的阻塞队列，且任务必须实现Runable接口，在JDK中提供了如下阻塞队列：
* 1、ArrayBlockingQueue：基于数组结构的有界阻塞队列，按FIFO排序任务；
* 2、LinkedBlockingQuene：基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene；
* 3、SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQuene；
* 4、priorityBlockingQuene：具有优先级的无界阻塞队列。
*/
this.workQueue = workQueue;
/**
* 线程池维护线程所允许的空闲时间。
* 当线程池中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了keepAliveTime；
*/
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
/**
* 线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：
* 1、AbortPolicy：直接抛出异常，默认策略；
* 2、CallerRunsPolicy：用调用者所在的线程来执行任务；
* 3、DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
* 4、DiscardPolicy：直接丢弃任务；
*/
this.handler = handler;
}
上面是ThreadPoolExecutor的构造方法与主要入参，我们可以结合线程池的任务调度图来理解相应字段代表的含义：
image.png
ThreadPoolExecutor提供的监控工具
threadPool.getActiveCount(); // 正在执行任务的线程数量
threadPool.getPoolSize()；// 当前线程数
threadPool.getTaskCount(); // 已执行与未执行的任务总数
threadPool.getCompletedTaskCount(); // 已完成的任务数
ThreadPoolExecutor源码解读
我们了解完上面关于线程池的基本概念后，下一步着手来对线程池的执行流程进行学习，首先我们先看下最基本的 execute()过程：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* 此方法一共分为三步
*
* 1. 当前线程数少于核心线程数，则新建一个线程放入池中，并把任务添加到该线程中
* 如果添加失败，则重新获取ctl值
*
* 2. 如果当前线程池是运行状态并且任务添加到队列成功，再次获取ctl值并判断线程池的运行状态。
* 如果不是运行状态，则需要移除该command并执行拒绝策略，如果当前工作线程数为0
* 就创建一个新的线程放入池中（此时command已经放入到workQueue啦）
*
* 3. 创建一个非核心线程，如果失败就执行拒绝策略
*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
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);
}
else if (!addWorker(command, false))
reject(command);
}
image.png
任务的执行流程梳理完毕后，我们接着看其中的 addWorker()方法：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
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;
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 {
// 把任务直接交给新创建的工作线程
w = new Worker(firstTask);
final Thread t = w.thread;
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());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// workers是一个HashSet
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
// 记录线程池中出现过的最大线程数量
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 线程开始工作
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
ThreadPoolExecutor.Worker类
线程池中的每一个线程被封装成一个Worker对象，ThreadPool维护的其实就是一组Worker对象，它继承了AQS，并实现了Runnable接口，注意其中的 firstTask和 thread属性：firstTask用来保存传入的任务；thread是在调用构造方法时通过ThreadFactory来创建的线程，是用来处理任务的线程。下面是它的构造与 run()方法：
Worker(Runnable firstTask) {
setState(-1); // 将state设置为-1是为了禁止在执行任务前对线程进行中断，因此，runWorker()方法中会先调用Worker对象的unlock方法将state设置为0
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
Worker继承了AQS，使用AQS来实现独占锁的功能。需要注意的是，Worker实现的 tryAcquire()方法是不允许重入的，而ReentrantLock是允许重入的。
在Worker构造中 setState(-1)的目的：
- lock方法一旦获取了独占锁，表示当前线程正在执行任务中；
如果正在执行任务，则不应该中断线程；- 如果该线程现在不是独占锁的状态，也就是空闲的状态，说明它没有在处理任务，这时可以对该线程进行中断；
- 线程池在执行 shutdown()或 tryTerminate()方法时会调用 interruptIdleWorkers()来中断空闲的线程，interruptIdleWorkers()会使用 tryLock() 来判断线程池中的线程是否是空闲状态；
- 之所以设置为不可重入，是因为我们不希望任务在调用像 setCorePoolSize()这样的线程池控制方法时重新获取锁。如果使用ReentrantLock，它是可重入的，这样如果在任务中调用了如 setCorePoolSize()这类线程池控制的方法，会中断正在运行的线程
我们接着看它的 runWorker()方法：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 响应构造方法的加锁，表示可以被中断了
// 是否因为异常退出循环
boolean completedAbruptly = true;
try {
// 如果task为空，则通过getTask来获取任务
while (task != null || (task = getTask()) != null) {
w.lock();
// 执行该if语句期间可能也执行了shutdownNow方法，shutdownNow方法会把状态设置为STOP
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task); // 空实现，可以被重写
Throwable thrown = null;
try {
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 {
// 如果task为null则跳出循环
processWorkerExit(w, completedAbruptly);
}
}
线程池随用随取的逻辑就在这个 runWorker()方法之中，它通过重写 run()方法，使其不断自旋并尝试去阻塞队列中获取任务。因此，它会一直处于运行或挂起状态而不会被销毁（当达到特定条件除外）。其中，getTask() 就是用来从阻塞队列中获取任务，代码如下：
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
/*
* 如果线程池状态rs >= SHUTDOWN，也就是非RUNNING状态，再进行以下判断：
* 1. rs >= STOP，线程池是否正在stop；
* 2. 阻塞队列是否为空。
* 如果以上条件满足，则将workerCount减1并返回null。
* 因为如果当前线程池状态的值是SHUTDOWN或以上时，不允许再向阻塞队列中添加任务。
*/
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// 是否需要进行工作线程的超时控制
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 获取前先判断条件
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 根据超时控制标识来使用获取任务的方法
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r; // 如果拿到任务，直接返回
timedOut = true; // 否则标识为超时
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
当 getTask() 返回null时，在 runWorker()中会跳出while循环，然后会执行 processWorkerExit()方法：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
// 根据线程池状态进行判断是否结束线程池
tryTerminate();
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);
}
}
至此，processWorkerExit()执行完之后，工作线程被销毁，以上就是整个工作线程的生命周期，从 execute()方法开始，Worker使用ThreadFactory创建新的工作线程，runWorker()通过 getTask()获取任务，然后执行任务，如果 getTask()返回 null，则进入 processWorkerExit()方法，整个线程结束，如图所示：
image.png
定时任务线程池ScheduledThreadPoolExecutor
ScheduledThreadPoolExecutor是一个可以处理周期性和延时性的任务，它继承了ThreadPoolExecutor，自身使用的阻塞队列是DelayedWorkQueue，其内部包含一个RunnableScheduledFuture数组，任务的插入基于最小堆排序，将最接近当前时间的任务放在数组的首个下标位置。
这里我们只看它重写的 take()方法，大概了解一下它是怎么实现周期执行的（读懂AQS，一切就很容易理解）：
public RunnableScheduledFuture<?> take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
RunnableScheduledFuture<?> first = queue[0];
if (first == null)
available.await();
else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
return finishPoll(first);
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}
ScheduledThreadPoolExecutor的任务调度流程如下图：
image.png
