深入理解线程池 - ThreadPoolExecutor_综合

线程是稀缺资源，创建线程，需要申请一定量的内存空间，具体申请多大主要取决于JVM虚拟机，由于对JVM不是很熟悉，后面有机会的时候了解一下，创建一个空线程，到底向虚拟机申请了多大的内存空间，如何通过命令查看，已经运行时如何查看。

使用线程池的目的：

线程是稀缺资源，不能频繁创建。
解耦作用，将线程的创建和运行分开，方便维护
对线程进行复用。

线程池肯定不是万能的，它适用于以下场景：

线程之间是独立的，不依赖于其他线程的执行时序，执行结果。
线程执行的任务在较短时间内可以完成
线程的创建数量具有不确定性，可能在某个时间创建多个线程。比如说服务器某个时间接受到多个客户端请求，客户端程序某一时间，向服务器请求多个图片等资源。

线程池构造函数

java 创建线程池最终都是调用ThreadPoolExecutor来创建的。ThreadPoolExecutor完整的构造函数如下

/** * corePoolSize - 核心工作线程大小 * maximumPoolSize - 最大工作线程大小 * keepAliveTime - 非核心工作线程空闲状态存活时间 * unit - 非核心工作线程空闲状态存活时间单位 * workQueue 任务阻塞队列 * 当线程池和任务阻塞队列都满后的拒绝策略 **/
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);

常用的任务阻塞队列有：

将Integer.MAX_VALUE 也认为是无界的。
ArrayBlockingQueue  使用数组来实现的有界的阻塞队列
LinkedBlockingQueue 使用单向链表实现的无界阻塞队列
LinkedBlockingQueue 使用双向链表实现的无界阻塞队列
PriorityBlockingQueue 支持优先级排序的无界阻塞队列
SynchronousQueue      同步阻塞队列，必须先将队列中的元素移出队列后，才能向队列添加元素

调用execute(Runnable command)线程池的执行情况：

当工作线程数小于核心工作线程数，则创建一个新的线程，执行入参的Runnable。
当工作线程数等于核心工作线程数，且阻塞队列未满，则将入参Runnable，添加到阻塞队列中。
当工作线程数等于核心工作线程数，且阻塞队列已满，且工作线程数小于最大工作线程数，则创建一个线程，执行入参的Runnable
当工作线程数等于核心工作线程数，且阻塞队列已满，且工作线程数不小于最大工作线程数，则执行拒绝策略

注意

只有在阻塞队列是有界大小的时候，参数 maximumPoolSize,keepAliveTime,unit,handler才是有意义的，如果阻塞队列是无界的，那么提交的Runnable只会添加到阻塞队列中，这样可能造成，提交的Runnable 长时间不运行。
在阻塞队列是有界的情况，当阻塞队列已满，且工作线程数小于最大工作线程数，这时候提交的线程是立马执行的，不会从阻塞队列中取出元素（Runnable）来执行，然后把当前提交的Runnable添加到队列中。
当非核心工作线程空闲时，当阻塞队列不为空，则会从阻塞队列中取出元素（Runnable）来执行。即只有当非核心工作线程空闲，且阻塞队列为空，这时候，非核心工作线程空闲时间才有效（参数keepAliveTime,unit）

线程池的相关状态

线程状态

由于线程池中使用到线程，线程的状态也会对线程池的状态进行影响，因此需要先了解线程的状态

NEW ，RUNNABLE ，TIMED WAITING ,WAITING,BLOCKED,TERMINATED

状态图转自：https://www.uml-diagrams.org/examples/java-6-thread-state-machine-diagram-example.html

在这里插入图片描述

线程池状态

线程池状态是使用 int 类型数据的高3位。源码如下

    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;private static final int TERMINATED =  3 << COUNT_BITS;

状态说明

RUNNING 运行状态，创建实例后，即为RUNNING状态

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// 初始状态为RUNNING

SHUTDOWN 关闭状态。不在接收新的Runnable,但当前运行的Runnable和阻塞队列中的Runnable会继续执行
STOP 指调用了 shutdownNow() 方法，不再接受新任务，同时抛弃阻塞队列里的所有任务并中断所有正在执行任务
TIDYING 所有任务都执行完毕，在调用 shutdown()/shutdownNow() 中都会尝试更新为这个状态。
TERMINATED 终止状态，当执行 terminated() 后会更新为这个状态

线程池执行流程

关于线程池是如何执行Runnable的，可以从以下几个问题分析：

线程池如何启动线程。
线程池如何复用线程
线程池如何销毁非核心线程

线程池核心工作线程启动执行任务流程

当工作线程数小于核心工作线程数，则创建一个新的线程，执行入参的Runnable。调用接口boolean addWorker(Runnable firstTask, boolean core)。此时，firstTask != null && core == true。

流程图如下
在这里插入图片描述
注意事项

在流程图中有多次检查当前线程池的状态的语句，这样做的原因是保存操作的时候，线程池在期望的状态中。
流程图中有两个判断一个判断是“工作线程数加1是否成功”，以及之后判断“判断当前线程的状态是否为RUNNING”，如果失败，则重新执行流程，这里的步骤是标准的乐观锁的流程。
如果添加成功，启动线程，注意线程的执行体在Worker中。

线程池中线程执行的任务 - Worker

线程池中线程执行的核心任务是Worker，掌握了Worker，就可以很清楚的了解，线程池中的线程是如何执行任务的。先上源码。

 private final class Worker extends AbstractQueuedSynchronizer implements Runnable{Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}public void run() {runWorker(this);}...}

Worker的构造函数完成了四件事件：

初始化了状态。
将execute(Runnable command)' 中的 command 保存在Worker中的 firstTask
通过线程制造工厂getThreadFactory()创建了线程，并保存在Worker中的 thread
当自己对象的应用，传递给新创建的线程中。

同时注意到Worker 本身也实现了Runnable 接口。因此调用Thread的start接口是，会调用到Worker中的run方法。
通过Worker的操作，可以理解为Worker 是对调用者传递进来Runnable进行了封装，让其拥有了线程，队列等属性。

addWorker

addWorker 入参说明

/** ** @param firstTask 调用者提交的任务（Runnable） * @param core 调用者提交的任务（Runnable）是否运行在核心工作线程 */
private boolean addWorker(Runnable firstTask, boolean core)

调用方式有三种：

1. firstTask ! = null && core == true 。将调用者提交的线程放置在核心工作线程执行。流程图参见上面
2. firstTask != null && core == false。将调用者提交的线程，放置在非核心工作线程执行，这里需要注意的是没有通过阻塞队列进行排队。
3. firstTask == null && core  == false 。通过循环从阻塞队列中获取任务，并在线程中执行

线程池中线程运行任务的核心runWorker

runWorker 是运行任务的核心，里面的代码详细的描述了线程的运行和线程的复用。源码如下

final void runWorker(Worker w) {// 因为是Worker本身也实现了Runnable，且对象自身的应用在构造函数的时候传递给了getThreadFactory() 创建的线程// 因此 Thread.currentThread() 和 w.thread 是等价的Thread wt = Thread.currentThread();// 获取到调用者提交的RunnableRunnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 如果是添加到核心工作线程执行的Runnable，此时task 肯定 不为nullwhile (task != null || (task = getTask()) != null) {w.lock();// 保证线程是可以执行的if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {beforeExecute(wt, task); // 执行任务前的hookThrowable thrown = null;try {task.run(); // 在线程中调用调用者提交Runnable中的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); // 任务完成后的hook}} finally {task = null; // 执行完成任务后，任务设置为nullw.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}}

线程池如何复用线程

线程池是如何复用线程的，核心代码还是runWorker，复用线程池，最为核心的代码是getTask。基本流程，可以参考注释。源码如下

	/*** 前提条件：函数运行在某一工作线程中。**/private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();// 获取线程池运行状态int rs = runStateOf(c);/*** 注意 RUNNIN 的值是最小的* 本质：状态要为 STOP,TIDYING,TERMINATED 即调用了shutdown() 或者 shutdownNow()* 并且 阻塞队列为空**/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 {/**poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则直到超时**/Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

线程池如何销毁非核心线程

在运行过程中，如果阻塞队列所有的Runnable被消费完成，才会销毁非核心工作线程。

final void runWorker(Worker w){try {// 如果阻塞队列为空，while条件不满足，跳出while循环。while (task != null || (task = getTask()) != null){}} finally {// 这个函数是销毁线程的函数processWorkerExit(w, completedAbruptly);}
}
// processWorkerExit 的代码简易如下
private void processWorkerExit(Worker w, boolean completedAbruptly){...// 这个函数本质是通过 Thread 的 interrupt 接口，对线程进行停止。tryTerminate();...
}