JDK17 线程池 ThreadPoolExecutor

线程池

线程池将创建线程和使用线程解耦。优点是

1. 避免重复创建和销毁线程，降低资源消耗。
2. 任务不用等待创建线程的时间，提高响应速度。
3. 统一管理线程，避免线程数量过多，造成OOM。保障系统稳定运行。

ThreadPoolExecutor

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; // 拒绝策略执行器
}

corePoolSize是核心线程数，默认核心线程不会被销毁。
maximumPoolSize 大等于corePoolSize，如果核心线程不足以任务并且任务队列已满，那么会新建非核心线程处理任务。如果线程数达到最大线程数，那么不执行任务，而是执行拒绝策略。非核心线程会被销毁，它们如果在keepAliveTime 存活时间内没有收到任务则会被销毁。
workQueue 任务队列存放工作线程还没来得及处理的任务。
threadFactory 是新建线程的工厂类，定义线程的名称，优先级以及是否是守护线程。默认用Executors.DefaultThreadFactory。
拒绝策略执行器是线程池无法接收新线程(可能原因1. 线程池已被关闭 2. 工作线程数已经是最大线程且任务队列满了)时执行的方法。默认用ThreadPoolExecutor#AbortPolicy.
最少只需要填四个参数，corePoolSize, maximumPoolSize, workQueue, keepAliveTime，就可以创建线程池。
这里一张图

状态

ctl表示线程池状态，它的类型是AtomicInteger ，AtomicInteger 类实现原子性操作int值。ctl的前3位是线程池状态，通过runStateOf()获得。ctl的后29位是运行数，workerCountOf()方法可得。
单独看状态，可以为分为5种。RUNNING, SHUTDOWN, STOP, TIDYING, TERMINATED。其中RUNNING<0，表示线程池正常运行。SHUTDOWN =0表示不接收新任务，但会执行完已有任务。STOP=1表示不接收新任务，且不执行已有任务。TIDYING=2表示线程池已清空。TERMINATED =3表示线程池已终止。
这里1张图。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 线程池状态
private static final int COUNT_BITS = Integer.SIZE - 3; // 29
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1; // 000(29个1)
private static final int RUNNING    = -1 << COUNT_BITS; // 111(29个0)
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 0
private static final int STOP       =  1 << COUNT_BITS; // 001(29个0)
private static final int TIDYING    =  2 << COUNT_BITS; // 010(29个0)
private static final int TERMINATED =  3 << COUNT_BITS; // 010(29个0)private static int runStateOf(int c)     { return c & ~COUNT_MASK; }
private static int workerCountOf(int c)  { return c & COUNT_MASK; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

向线程池添加任务 execute

public void execute(Runnable command) { // command是线程池接收的任务if (command == null) // 参数校验throw new NullPointerException();int c = ctl.get(); // 线程池状态if (workerCountOf(c) < corePoolSize) { // 如果工作线程数少于核心线程数if (addWorker(command, true)) // 增加核心工作线程return;c = ctl.get();}// 运行至此，当前线程数不少于核心线程数// isRunning()返回线程池是否是RUNNING状态，是则true// workQueue.offer(command)) 将任务阻塞地加入队列if (isRunning(c) && workQueue.offer(command)) {// 运行至此，任务已进入阻塞队列int recheck = ctl.get(); // 如果线程池状态不是RUNNING，取消任务，执行拒绝策略if (! isRunning(recheck) && remove(command))reject(command);// 如果工作线程数是0，启动一个非核心工作线程else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 运行至此，没有成功加入阻塞队列// 添加非核心线程执行任务else if (!addWorker(command, false))reject(command);
}

addWorker(command, true)表示线程池增加核心工作线程，addWorker(command, false)表示增加非核心线程。execute()方法的流程图如下
直接上图。

Worker

想理解addWorker()方法，得先理解Worker类。Worker封装Thread类型变量和Runnable类型变量。this.thread = getThreadFactory().newThread(this);表示thread是工作线程，执行Worker类的run()方法，后面addWork()方法会执行worker.start()真正开启新线程。后者表示任务。
Worker类继承AbstractQueuedSynchronizer是为了实现不可重入独占锁， state是Worker类的状态，表示当前工作线程是否正在执行任务。除了工作线程执行线程池任务，调用线程池的多个线程可能也会操作线程池任务(比如STOP状态中断所有任务)，为了防止线程安全问题，因此加锁。

private final class Workerextends AbstractQueuedSynchronizerimplements Runnable
{final Thread thread; // 工作线程Runnable firstTask; // 线程池任务Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}public void run() {runWorker(this);}
}  

runWorker(Worker w)定义工作线程声明周期内的操作。只有task.run();表示工作线程执行客户端发送给线程池的任务，其余的源码表示工作线程的其他操作。三层try-catch，第一层表示工作线程从任务队列拉取任务。第二层表示工作线程执行环境任务beforeExecute(), afterExecute()。第三层表示工作线程执行任务。

final void runWorker(Worker w) {Thread wt = Thread.currentThread(); // 当前线程是工作线程，不是调用线程池的客户端线程Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;// 第一层try结束代表当前工作线程要被销毁try {// task != null 表示新建工作线程时就分配任务// getTask() 表示阻塞地从任务队列拉取任务while (task != null || (task = getTask()) != null) {w.lock();// 至此，工作线程获得不可重入独占锁// (runStateAtLeast(ctl.get(), STOP)表示线程池的状态是`STOP,TIDYING,TERMINATED`，线程池要求中断任务，中断任务是指设置中断位为true，而不是强制终止任务// 这个if语句的语义是：清除当前线程的中断标记，不干扰执行任务期间的中断状态，除非如果线程池要求中断任务。// 尽可能获取最新线程池状态，因此执行2次(runStateAtLeast(ctl.get(), STOP)if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();// 运行至此，正常情况下线程中断位为false，线程池状态为`STOP,TIDYING,TERMINATED`时中断位为true// 第二层try表示工作线程执行本次任务try {beforeExecute(wt, task); // 钩子方法，所有任务公用一个方法体// 第三层try表示执行客户端提交给线程池的任务// `run()`方法不会开启一个新线程try {task.run();afterExecute(task, null);} catch (Throwable ex) {afterExecute(task, ex);throw ex;}} finally {task = null;w.completedTasks++;w.unlock();}}// 运行至此，代表任务没有抛出异常completedAbruptly = false;} finally {// 运行至此，completedAbruptly =true表示任务`task.run()`抛出异常。// 运行至此，工作线程没有获取到任务，如果不符合线程池要求，则销毁这个工作线程processWorkerExit(w, completedAbruptly);}// 至此，工作线程执行完`run()`方法中所有内容，终止该线程
}

工作线程的终结方法processWorkerExit()。尝试销毁工作线程，并且尝试终止线程池。
decrementWorkerCount()方法就是将状态ctl-1，表示工作线程减1.如果任务没有抛出异常，工作线程会在getTask()方法里执行该方法。
mainLock是线程池的重入独占锁，保证多个线程调用线程池时线程安全。

private void processWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusteddecrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks; // 线程池完成的任务数+=当前工作线程完成的任务数workers.remove(w); // 当前工作线程被移出集合，意味着工作线程被销毁} finally {mainLock.unlock();}tryTerminate(); // 当前线程已被销毁，是否意味着线程池进入非`RUNNING`状态。int c = ctl.get();if (runStateLessThan(c, STOP)) { // true表示：线程池的状态是`RUNNING, SHUTDOWN`，继续执行线程池已有任务if (!completedAbruptly) { // 如果任务没有抛出异常int min = allowCoreThreadTimeOut ? 0 : corePoolSize; // 线程池最小线程数if (min == 0 && ! workQueue.isEmpty()) // 如果最小线程数为0，并且仍要执行任务min = 1; // 最小线程为1if (workerCountOf(c) >= min) // 当前工作线程满足执行任务的要求return; // replacement not needed}// 如果任务抛出异常或者当前工作线程不满足执行任务的要求，新建一个非核心线程，不布置任务addWorker(null, false);}}

tryTerminate()方法尝试让线程池进入TERMINATED状态。interruptIdleWorkers()方法将中断一个空闲的工作线程。

final void tryTerminate() {for (;;) { // 自旋int c = ctl.get();if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateLessThan(c, STOP) && ! workQueue.isEmpty()))// 运行至此，线程池状态要么是`RUNNING`，要么是`TIDYING, TERMINATED`，要么是`SHUTDOWN`并且任务队列为空// `return;`表示第一种情况和第三种情况下线程池还不能进入终止状态，第二种情况表示不用重复执行下述操作return;// 运行至此，线程池状态不是`RUNNING, TIDYING, TERMINATED`。线程池的状态可能是//    1. `SHUTDOWN`并且任务队列为空//    2. `STOP`，无所谓队列空不空，因为工作线程不执行队列中任务// 第一种情况下可能有非常多工作线程阻塞在`getTask()`方法的`workQueue.take()`/`workQueue.poll()`方法。设置工作线程的标记位后，工作线程会跳出阻塞抛出`InterruptedException`异常。为了防止大规模抛出异常，造成系统不稳定，因此一个一个中断线程。if (workerCountOf(c) != 0) { // Eligible to terminateinterruptIdleWorkers(ONLY_ONE); // 中断一个工作线程return;}// 至此，`SHUTDOWN`满足任务队列为空并且工作线程数为0，`STOP`满足工作线程数为0final ReentrantLock mainLock = this.mainLock;mainLock.lock(); // 获取线程池的重入独占锁try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { // true则表示当前线程池状态已经被修改为`TIDYING`，并且工作线程数是0 只有1个线程能返回truetry {terminated(); // 钩子方法，线程池终止方法} finally {ctl.set(ctlOf(TERMINATED, 0));// 至此，终止线程池termination.signalAll();}return;}} finally {mainLock.unlock();}// else retry on failed CAS}
}

线程池新建工作线程 addWorker

Worker类定义了工作线程的内容。但是创建工作线程是线程池调用addWorker()方法完成的。retry循环尝试获取新增线程授权，获得授权后新增工作线程。

private boolean addWorker(Runnable firstTask, boolean core) {retry:for (int c = ctl.get();;) {// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN) // 表示线程池非`RUNNING`&& (runStateAtLeast(c, STOP) // 表示线程池为`STOP, TIDYING, TERMINATED`|| firstTask != null // 任务不为空|| workQueue.isEmpty())) // 任务队列为空return false;// 运行至此，没有`return false`，可能情况//     1. 线程池状态为`RUNNING`//     2. 线程池状态为`SHUTDOWN`，并且以下条件满足1个//          1. 任务为空，表示不接收新任务//          2. 任务队列非空，表示仍然执行任队列里的任务for (;;) { // 自旋if (workerCountOf(c)>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))return false; // 工作线程数已经达到阈值了，不创建新线程if (compareAndIncrementWorkerCount(c)) break retry; // 成功使状态`ctl`+1，获得创建新线程许可，跳出循环// 至此，没有获得许可，继续尝试c = ctl.get();  // Re-read ctlif (runStateAtLeast(c, SHUTDOWN))continue retry; // 如果线程池状态不是`RUNNING`，那么重新进入retry循环。如果是`RUNNING`,这个for循环自旋就行。// else CAS failed due to workerCount change; retry inner loop}}boolean workerStarted = false; // 工作线程是否执行`start()`方法，即新建线程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(); // 获取线程池独占锁，为了原子性更新`workers`,`largestPoolSize`。try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();if (isRunning(c) ||(runStateLessThan(c, STOP) && firstTask == null)) {// 再次判断线程池状态，可能情况//    1. `RUNNING`//    2. `SHUTDOWN`并且任务为空if (t.getState() != Thread.State.NEW)throw new IllegalThreadStateException(); // 线程状态已被篡改，抛出异常workers.add(w);workerAdded = true;int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;}} finally {mainLock.unlock();}if (workerAdded) { // 如果工作线程进入集合，表示线程池已经收录这个`Worker`对象，则可以启动线程t.start(); // 创建新线程workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;
}

如果没有新建工作线程，则执行回退。

private void addWorkerFailed(Worker w) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {if (w != null)workers.remove(w); // 线程集合删除本工作线程decrementWorkerCount(); // 减少工作线程数tryTerminate(); // 检查线程池是否已经准备进入终止状态} finally {mainLock.unlock();}
}

拒绝策略

JDK17提供四种拒绝策略。可以自定义拒绝策略，定义RejectedExecutionHandler 接口的实现类，重写rejectedExecution()方法。ThreadPoolExecutor类的默认拒绝策略是AbortPolicy。

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

// AbortPolicy 直接抛出异常
public static class AbortPolicy implements RejectedExecutionHandler {public AbortPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {throw new RejectedExecutionException("Task " + r.toString() +" rejected from " +e.toString());}
}
// DiscardPolicy丢弃，不做任何处理
public static class DiscardPolicy implements RejectedExecutionHandler {public DiscardPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}
}
// DiscardOldestPolicy尝试丢弃任务队列的头部线程
public static class DiscardOldestPolicy implements RejectedExecutionHandler {public DiscardOldestPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {e.getQueue().poll();e.execute(r);}}
}
// CallerRunsPolicy尝试使用线程池对象的线程处理任务，也就是谁提交任务，谁处理任务。run()方法不会新建线程对象，而是直接本线程执行任务内容
public static class CallerRunsPolicy implements RejectedExecutionHandler {public CallerRunsPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {r.run();}}
}