线程池执行原理
目录
线程池的基本介绍
线程池参数
线程池状态及基本方法
线程池的执行过程
内置线程池
newFixedThreadPool
newSingleThreadExecutor
newCachedThreadPool
newScheduledThreadPool
线程池的基本介绍 线程池在项目中的出场率还是比较高的，如果以比较白话的语言介绍线程池的话，其就是一个线程集合+任务队列，当有某个任务请求到达线程池时，塞入任务队列，然后线程无限循环从队列中取任务这样一个简单场景。那我们为什么需要线程池呢？
- 避免线程频繁的创建和销毁，减少资源的消耗
- 通过线程池可以更好的管理线程资源，如分配、调优和监控
- 提高响应速度，当任务来临时无需等待线程创建即可响应
public ThreadPoolExecutor(int corePoolSize, // 核心线程数量
int maximumPoolSize, // 线程池最大线程数量
long keepAliveTime, // 非核心线程数存活时间
TimeUnit unit, // 存活时间单位
BlockingQueue 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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
} 下面稍微详细点介绍这几个参数，然后再结合线程池的执行过程源码分析其各参数的使用
int corePoolSize：线程池中的核心线程数量，即使没有任务，这些线程也不会销毁。
int maximumPoolSize：线程池中的最大线程数量，即线程池所支持创建的最大数量线程，即使任务超级多，也只会有 maximumPoolSize 数量的线程在运行。
long keepAliveTime：非核心线程的存活时间，当任务数量超过核心线程数量时，只要 corePoolSize < maximumPoolSize，线程池便会创建对应的线程数去执行任务，当线程池中存活的线程数量大于核心线程时，如果等了 keepAliveTime 时间仍然没有任务进来，则线程池会回收这些线程。
TimeUnit unit：非核心线程存活时间的具体单位，即等待多少毫秒、秒等。
BlockingQueue workQueue：存储线程任务所用的队列，提交的任务将会被放到该队列中。
ThreadFactory threadFactory：线程工厂，主要用来创建线程的时候给线程取名用，默认是pool-1-thread-3
RejectedExecutionHandler handler：线程拒绝策略，当存储任务所用的队列都被填满时，新来的任务此时无处存放，那么需要提供一种策略去解决这种情况。 线程池状态及基本方法 该类定义了一些线程池的基本状态及基本方法，在阅读源码前需要先大概知晓：
/*
* 该对象高3位维护线程池运行状态，低29位维护当前线程池数量
*/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
/*
* 该值等于29，用于左移
*/
private static final int COUNT_BITS = Integer.SIZE - 3;
/*
* 线程池支持的最大线程数量，即29位所能表示的最大数值，2^29 - 1
*/
private static final int CAPACITY= (1 << COUNT_BITS) - 1;
/*
* 即高3位为111，低29位为0，该状态的线程池会接收新任务，并且处理阻塞队列中正在等待的任务
*/
private static final int RUNNING= -1 << COUNT_BITS;
/*
* 即高3位为000，不接受新任务，但仍会处理阻塞队列中正在等待的任务
*/
private static final int SHUTDOWN=0 << COUNT_BITS;
/*
* 高3位为001，不接受新任务也不处理阻塞队列中的任务
*/
private static final int STOP=1 << COUNT_BITS;
/*
* 高3位为010，所有任务都被终止了，workerCount为0，为此状态时还将调用terminated()方法
*/
private static final int TIDYING=2 << COUNT_BITS;
/*
* 高3位为011，terminated()方法调用完成后变成此状态
*/
private static final int TERMINATED =3 << COUNT_BITS;
这些状态均由int型表示，大小关系为 RUNNING
/*
* c & 高3位为1，低29位为0的~CAPACITY，用于获取高3位保存的线程池状态
*/
private static int runStateOf(int c){ return c & ~CAPACITY;
}
/*
* c & 高3位为0，低29位为1的CAPACITY，用于获取低29位的线程数量
*/
private static int workerCountOf(int c){ return c & CAPACITY;
}
/*
* 参数rs表示runState，参数wc表示workerCount，即根据runState和workerCount打包合并成ctl
*/
private static int ctlOf(int rs, int wc) { return rs | wc;
}线程池的执行过程 具体执行逻辑主要在 ThreadPoolExecutor 的 execute（） 方法，源码如下：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// ctl对象功能很强大，其高3位代表线程池的状态，低29位代表线程池中的线程数量
int c = ctl.get();
// 如果当前线程池中线程数量小于核心线程数，则新建一个线程并将当前任务直接赋予该线程执行
if (workerCountOf(c) < corePoolSize) {
// 如果新建线程成功则直接返回
if (addWorker(command, true))
return;
// 到这一步说明新建失败，可能是线程池意外关闭或者是由于并发的原因导致当前线程数大于等于核心线程数了，重新获取ctl对象
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);
}
// 如果处于非运行态或者入队列不成功（队列满了），尝试扩容线程池线程数量至maxPoolSize，若扩容失败，则拒绝该任务
else if (!addWorker(command, false))
reject(command);
}简单总结一下，主要分为三步：
- 先判断当前运行的线程数是否小于核心线程数，若是则新建一个线程来处理该任务，处理成功则直接返回，否则继续判断
- 判断当前线程池是否处于运行态且任务入队列是否成功，若是则进行双重检查，确保池中仍有运行的线程数
- 若第二步入队列失败，线程池尝试扩容至最大线程数，若失败则拒绝该任务
下面我们来看看新建线程的方法 addWorker（）
private boolean addWorker(Runnable firstTask, boolean core) {
// 首先是一个外层死循环
retry:
for (;
;
) {
int c = ctl.get();
int rs = runStateOf(c);
// 检查当前线程池是否处于非运行态，同时确保队列中任务数不为空
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内存死循环修改运行的线程数量
for (;
;
) {
int wc = workerCountOf(c);
// core参数确保不会超过线程池设定的值
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 采用CAS算法将线程数+1，如果成功则直接跳出外循环，失败主要是因为并发修改导致，那么则再次内循环判断
if (compareAndIncrementWorkerCount(c))
break retry;
// 确保线程池运行状态没变，若发生改变，则从外循环开始判断
c = ctl.get();
if (runStateOf(c) != rs)
continue retry;
}
}boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 新建Worker内部类时主要干了两件事，一个是设置AQS同步锁标识为-1，另一个是调用线程工厂创建线程并赋值给Worker的成员变量
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
// 在往workers这个线程集合增加线程时需要进行加锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 此处需要进行二次检查，防止线程池被关闭等异常情况
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// 如果当前线程已经启动，处于活跃态则抛异常
if (t.isAlive())
throw new IllegalThreadStateException();
// workers是一个HashSet集合
workers.add(w);
int s = workers.size();
// 设置最大池大小，同时标识任务线程增加成功，即 workerAdded 设为true
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 如果任务线程成功增加，则在此处启动线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}看似很长其实都是在做一些基本的判断，防止异常情况，简单总结下：
- 首先判断是否处于运行态且当前运行的线程数是否小于上限值，若是则通过CAS算法先将运行线程数+1
- CAS操作成功后，新建Worker线程并通过加锁的方式将其加入到线程集合，然后启动线程
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** 由线程工厂所创建的线程对象 */
final Thread thread;
/** 业务任务对象 */
Runnable firstTask;
/** 当前线程所执行任务的计数 */
volatile long completedTasks;
/**
* 初始化方法，主要是设置AQS的同步状态private volatile int state，是一个计数器，大于0代表锁已经被获取，设为-1后即禁止中断
*/
Worker(Runnable firstTask) {
setState(-1);
// 将lock标识设为-1，
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}public void run() {
runWorker(this);
}/**
* 下面的都是与锁相关的方法，state为0代表锁未被获取，1代表锁已经被获取
*/
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;
// 控制中断主要就是体现在中断前会判断 getState() >= 0
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}Worker类本身实现了Runnable，又继承了AbstractQueuedSynchronizer，所以其既是一个可执行的任务，又可以达到锁的效果，注意该锁是一个简单的不可重入锁
最后看下线程池中主要执行业务任务的方法 runWorker（）：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 将state置为0，允许线程中断
w.unlock();
boolean completedAbruptly = true;
try {
// task为上层透传进来的指定业务线程，若为空则循环通过getTask()获取任务执行
while (task != null || (task = getTask()) != null) {
// 这里的加锁不是为了防止并发，而是为了在shutdown()时不终止正在运行的任务
w.lock();
// 双重检查防止线程池状态不大于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 {
// 处理业务线程抛异常的情况
processWorkerExit(w, completedAbruptly);
}
}内置线程池 jdk为我们内置了四种线程池，均是通过 Executors 工厂类创建，不是很建议通过这种方式创建线程池，下面会介绍为什么不。
newFixedThreadPool
这是一个定长线程池
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue());
} 顾名思义创建一个固定长度的线程池，该线程池核心线程数与最大线程数是一致的，因此非核心线程存活时间参数也就没什么意义了，其最大问题就是选用的阻塞队列是无界的，若任务无穷多便会一直加加加，最终将内存撑爆，一般用于负载较重的服务器，需要限制指定的线程数。
newSingleThreadExecutor
这是一个单线程化的线程池
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue()));
} 看参数也能发现其只初始化了一个线程，自始至终池里也只有一个线程在工作，听起来可真可怜，唯一的问题也是阻塞队列是无界的，可能会将内存撑爆，一般用于需要严格控制任务执行顺序的场景。
newCachedThreadPool
这是一个可缓存线程池
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue());
} 核心线程数为0，最大线程数趋近于无限大，即全部采用非核心线程来处理任务，每个线程的存活时间则为60s，注意这里使用到了同步队列，即只会传递任务不会保存任务，一旦有任务来则查看是否有空闲线程，若无则新建一个线程，当任务请求的速率大于线程处理的速率，那么线程数会越来越多，最终将内存撑爆，一般用于并发执行大量短期的小任务，由于空闲60s的线程会被回收，因此长时间保持空闲的该线程池不会占用任何资源。
newScheduledThreadPool
这同样是一个定长线程池，但它支持定时及周期性任务执行
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}可以看到其最大线程数依旧无限大，因此一定程度上也会存在内存撑爆问题，在队列使用上则选择了延迟阻塞队列 DelayedWorkQueue，具体执行任务时也比较复杂。
【线程池执行原理】
推荐阅读
- CVE-2020-16898|CVE-2020-16898 TCP/IP远程代码执行漏洞
- Linux下面如何查看tomcat已经使用多少线程
- 多线程NSOperation
- 字符串拼接成段落，换行符（\n）如何只执行n-1次
- 欢乐小分队内蒙东北行第六站（第十二天）五大连池印象之（奇特壮观的火山地貌景观）
- 寻美姬渐不遇
- 墨池寻欢，书海探险（6）（临赵松雪《心经》）
- springboot整合数据库连接池-->druid
- spring|spring boot中设置异步请求默认使用的线程池
- R语言|R语言 函数