vlambda博客
学习文章列表

概要

线程池，大家都很熟悉了，我们在平时应用中也用的很多。对线程池，ThreadPoolExecutor 的实现原理有一定理解后，我们可以把它用的更好，对它的参数有更加深刻的理解，甚至我们可以扩展，监控自己的线程池。

ThreadPoolExecutor实现原理

本文代码基于JDK1.8

线程池相关的类的关系

我们先看看主要的类ThreadPoolExecutor的继承关系

平时可能还有用到的 Executors 类，这是一个工具类，提供 newFixedThreadPool newCachedThreadPool newScheduledThreadPool 等静态方法方便我们创建线程池，最终还是调用 ThreadPoolExecutor 来创建的，一般规范不建议直接使用 Executors 来创建线程池。

线程池创建的主流程

线程池的状态

先看看线程池的状态有哪些，对它有初步的理解

线程池的状态和运行的线程数只用了一个 int，其中高3位表示状态，低29位表示线程数。状态表示的意思和状态之间的转换：RUNNING - 可以接受新的任务，及执行队列中的任务

SHUTDOWN - 不接受新的任务，但会继续执行队列中的任务

STOP - 不接受新的任务，既不会执行队列中的任务，还会中断执行中的任务

TIDYING - 全部任务已经终止，且线程数为0

TERMINATED - 线程池完全终止

RUNNING -> SHUTDOWN - 执行了 shutdown()

(RUNNING or SHUTDOWN) -> STOP - 执行了shutdownNow()

SHUTDOWN -> TIDYING - 队列和线程为空

STOP -> TIDYING - 线程为空

TIDYING -> TERMINATED - terminated() 这个勾子方法执行完毕

线程池的创建

ThreadPoolExecutor 的构造函数

线程池的创建只是初始化了一些参数，但理解好这些参数对我们使用线程池很有帮助。corePoolSize - 核心线程数，除非 allowCoreThreadTimeOut 为 true（默认false），否则即使没有任务，也会维持这么多线程。

maximumPoolSize - 最大线程数，corePoolSize 满了的话，会把任务放到队列，如果队列满了的话（假设队列有界），就会继续创建线程直到 maximumPoolSize，如果超过 maximumPoolSize 则会执行 reject 策略。

workQueue - 用来存放任务的队列，是一个 BlockingQueue，常用的有 LinkedBlockingQueue，ArrayBlockingQueue，SynchronousQueue。

keepAliveTime - 空闲线程的存活时间，如果当前线程数 > corePoolSize，且某个线程空闲了这么多时间（没有获取到任务并运行），那么这个线程会被 remove 掉。

unit - keepAliveTime 的单位，内部会统一转换成 nanos

threadFactory - 用来创建线程的 ThreadFactory，主要用来给线程命名（方便查看日志），设置 daemon，优先级和 group 等，Executors 有 DefaultThreadFactory 这个默认实现。

handler - reject 具体执行策略，reject 的条件上面已经说了，一般内置有以下几个，也可以自己实现

CallerRunsPolicy - 不使用线程池的线程执行该任务，直接用当前执行任务（例如 main 线程）

AbortPolicy - 直接抛异常

DiscardPolicy - 无视不做处理，相当于抛弃掉

DiscardOldestPolicy - 将队列头的任务取出来抛弃掉，然后运行当前任务

线程池的执行

一般我们使用 ExecutorService 的 submit 方法来使用线程池执行一个任务，这个方法调用到 AbstractExecutorService

这里我们看到所有 task 无论是 Callable 还是 Runnable 的都会包装成一个 RunnableFuture 也就是 FutureTask，返回给我们。

execute方法

重点看 execute 方法，调用了 ThreadPoolExecutor 的 execute

我们分三种情形来看，每个是一个 if 条件：第一，当 workCount < corePoolSize 时，直接创建 worker 线程；

第二，如果上面创建失败（可能是线程池正在处于关闭状态，可能是 workCount > corePoolSize 了 - 并发场景），那么这时把任务放入 workQueue 队列；下面的判断是用来防止，线程池不在运行了，就把任务删掉；如果没有线程了就加一个；

第三，来到这步说明上面放队列不成功（可能是队列是有界的，满了），那么就继续创建线程来满足，如果这都创建失败（可能是 > maximumPoolSize）就 reject 了；

addWorker方法

继续看看重点的 addWorker 方法，addWorker 分开两部分来看。

这一步是判断是否可以增加 worker 的重点：

第一，首先开始的判断有点奇怪，我也不是很明白，先认为它是如果状态是 SHUTDOWN 则不允许创建线程；

第二，下面有个 core 参数，true 使用 corePoolSize，false 使用 maximumPoolSize，我们上面说的 execute 方法第一次就是传 true 的，第二次就传 false。所以这里就是对 size 做判断， 如果 >= size 则返回 false，否则 cas 一下，成功了就 break 执行下面的代码；

第三，如果 cas 失败，说明有其他并发竞争，则 cintinue 循环上面的步骤判断。

来到这一步说明可以创建 worker 了，这里用了一个全局 lock，来控制创建线程和关闭线程的不能同时做。可以看到步骤就是 new 一个 worker，add 到 workers 里，workers 是一个 HashSet。

largestPoolSize 来用记录最大的线程数。如果 workerStarted == false（线程池在关闭或创建 worker 时异常），则会 addWorkerFailed 方法。

主要就是 remove 掉 worker，扣减计数，这里还会调用 tryTerminate 。这个方法会在很多地方用到，它的作用就是防止线程池在终止状态这种情形，去终止线程。

Worker是什么

我们刚刚一直说 worker，那到底 Worker 究竟是什么？我们现在来看看

我们可以看到 Worker 是一个 AQS 和 Runnable。为什么是一个 AQS ？我们可以结合注释和代码可以得到，worker 在跑任务的时候会 lock 住，在被中断时会 tryLock，利用上锁这一点来区分这个 worker 是否空闲。Worker 中重写 AQS 的方法。（感概：AQS 真是个简单易用，用于并发控制的好工具！）

为什么是一个 Runnable ？我们看看 Worker 的构造函数，在创建 Thread 时把自己 this 传到 thread 的参数，说明 worker 的 thread 跑的是自己，这时我们就知道 worker 的入口了。

Worker 的 run 方法

runWorker方法

重点的 runWorker 方法

task 可能是传进来的 firstTask 或者 getTask() 获取到的，getTask 也是重点方法，等下讲到；运行 task 时会上锁，锁的作用我刚刚已经说了；如果线程池状态是 STOP 则中断线程；这里放了两个勾子 beforeExecute 和 afterExecute 方法来提供给子类做点事情，一般用于监控或统计线程池的执行情况；执行任务就直接 task.run() 了，还记得我说过这个 task 是一个 FutureTask，如果run 的时候抛出异常，FutureTask 会 catch 掉不会再 throw（如果大家对 FutureTask 不熟悉就先这样理解），所以这里不会进入 catch，也就是不会 throw x 了。如果 task 不像 FutureTask 一样自己处理掉异常，那就会 throw x 了，那么 worker 的线程就会跳出 while 循环，完成使命，结束自己；获取不到 task （task 为null）或者循环过程中异常，最后都会执行 processWorkerExit。

processWorkerExit 的作用主要就是 remove 掉 worker，那么扣减 workCount 呢？好像没有看到。这里用了 completedAbruptly 这一变量来控制是否在 processWorkerExit 扣减 workCount，因为有可能是在 getTask 已经扣减了，那么在 processWorkerExit 就不用重复扣减。我们结合 runWorker 来看看，分两种情况：

1、如果 firstTask 为 null，那么会走到 getTask 方法，如果 getTask 返回 null，那么说明已经是扣减了，这时退出循环，completedAbruptly = false，不用重复扣减。

2、如果 firstTask 不为 null （1）执行 firstTask 正常结束，然后循环，走到 getTask，如果返回 task 为 null，那么 completedAbruptly = false，不用重复扣减。

（2）执行 firstTask 执行异常，这时 completedAbruptly = true，需要扣减 这里我们又看到 tryTerminate 了；下面的判断主要是尝试去增加一个 worker，因为你 remove 掉了一个，如果条件允许，那就加回一个呗。

getTask方法

看看重点的 getTask 方法

在 getTask 时如果发现时线程池在关闭状态，那么就需要停止获取任务了；

如果 wc > maximumPoolSize，超过了最大 size 了，就去 cas 扣减 workCount 一下，成功就返回 null；

如果 wc > corePoolSize（小于 maximumPoolSize），且 timedOut 的话，说明这个 worker 也有点“多余”，也去扣减 workCount。注意这里对 timedOut 的判断，通过 queue 的定时 poll 方法，时间是线程池的构造参数 keepAliveTime，如果过了这么久都还没获取 task，说明 queue 是空的，有空闲的线程，那就把这个 worker remove 掉吧；

如果 wc < corePoolSize 的话，那就调用 queue 的 take 方法一直阻塞获取 task；

还有 allowCoreThreadTimeOut 参数，它的意思是忽略对 corePoolSize 的判断。

关闭线程池

上面已经把线程的创建和执行基本说得7788了，我们看看关闭线程池是如何做的，其实在上面的很多方法中，都看到很多对如 SHUTDOWN 的判断了。主要有 shutdown 和 shutdownNow 这两个方法。

这两个方法很相似，从名字来看 shutdown 显得更加的柔性，实际也是。shutdown – 不接受新的 task，在运行和已经在队列的 task 可以继续运行；把状态改为 SHUTDOWN；中断空闲 worker，这个在上面已经提到过了，用锁对是否空闲做判断。

interruptIdleWorkers 打断空闲的线程

这里还有个 onShutdown 勾子方法。

shutdownNow – 不接受新的 task，中断已经在运行的线程，清空队列；把状态改为 STOP；强制中断所有在运行 worker 的线程；drainQueue，相当于把队列的 task 丢弃掉；

总结

线程池ThreadPoolExecutor实现的原理，就是用 Worker 线程不停得取出队列中的任务来执行，根据参数对任务队列和 Workers 做限制，回收，调整。

参考资料