线程的复用---线程池原理解析
线程的复用---线程池原理解析
一：简述
本文基于java11对线程池的参数，执行任务的流程以及原理进行解析，并且对线程池关键性源码进行了分析。
二：线程池的参数
1. corePoolSize： 用于指定线程池的核心线程数量
2. maximumPoolSize： 用于定义线程的最大线程数
3. keepAliveTime： 非核心线程的空闲时间
4. timeUnit： 空闲时间单位
5. blockingQueue： 阻塞队列
6. threadFactory： 线程工厂，可以用于定义线程的名称，线程的优先级，是否是守护线程等信息，支持自定义线程工厂，默认采用DefaultThreadFactory
7. rejectedExecutionHandler： 拒绝策略，jdk已经实现了四种拒绝策略
a. AbortPolicy 直接抛出异常，也是默认的拒绝策略
b. CallerRunsPolicy 使用当前线程执行任务
c. DiscardOldestPolicy 丢弃最早加入的任务，并执行任务
d. DiscardPolicy 直接丢弃当前任务
三：Java提供的四种线程池
在Executors工厂类中，Java默认提供了四种类型的线程池。
1.newFixedThreadPool：有固定的线程数的线程池，最大线程数等于核心线程数，阻塞队列使用LinkedBlockingQueue，线程空闲时间为0
2.newCachedThreadPool：可缓存的线程的线程池，核心线程数为0，最大线程数为Integer.MAX，线程空闲时间为60s，阻塞队列使用SynchronousQueue
3.newScheduledThreadPool：能够定时完成任务的线程池，核心线程数自定义，最大线程数为Integer.MAX，线程空闲时间为0，阻塞队列为DelayedWorkQueue
4.newSingleThreadExecutor：单线程的线程池，核心线程数和最大线程数都为1，阻塞队列为LinkedBlockingQueue，线程空闲时间为0
四：线程池状态
//运行状态
private static final int RUNNING = -1 << COUNT_BITS;
// shutdown状态 不接收新任务，但能处理已添加的任务
private static final int SHUTDOWN = 0 << COUNT_BITS;
//停止状态 不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。
private static final int STOP = 1 << COUNT_BITS;
//当线程池在SHUTDOWN状态下，阻塞队列为空并且线程池中执行的任务也为空时 会由SHUTDOWN状态扭转为TIDYING
private static final int TIDYING = 2 << COUNT_BITS;
//线程池彻底终止，就变成TERMINATED状态
private static final int TERMINATED = 3 << COUNT_BITS;五：线程池执行任务流程图
六：源码解析
ThreadPoolExecutor threadPool = new
ThreadPoolExecutor(4, 8, 60,
TimeUnit.SECONDS, new ArrayBlockingQueue<>(100),new ThreadPoolExecutor.AbortPolicy());
threadPool.execute(()->{
System.out.println("线程池原理解析");
});execute()方法
execute()方法作为阅读相关源码的入口，首先对execute()方法进行分析，从execute()方法可以看出线程池的主流程，工作线程数量没有达到核心线程数的话会优先创建线程并执行任务，达到核心线程数量之后，将任务放入到阻塞队列中，阻塞队列满了之后继续创建线程执行任务，直到工作线程数量达到最大线程数，再执行拒绝策略。其中包含两个比较重要的方法 addWorker()，reject()方法。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//c的前3位表示线程池的状态 后29位表示线程的数量
int c = ctl.get();
//如果当前线程数量小于核心线程数
if (workerCountOf(c) < corePoolSize) {
//调用addWorker添加一个线程并执行任务
//addWorker在线程池状态不是running 或者 工作线程数量超过限制返回false
if (addWorker(command, true))
return;
c = ctl.get();
}
//走到下面的逻辑 addworker是添加非核心线程了 也就是第二个参数为false
//如果线程池是运行状态 通过offer方法把任务放入阻塞队列
//offer方法放入队列成功返回true 失败返回false
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//进一步判线程池状态 不是运行状态 而且移除任务成功
if (! isRunning(recheck) && remove(command))
//执行拒绝策略
reject(command);
else if (workerCountOf(recheck) == 0)
//如果工作线程数量为0 那么重新添加非核心工作线程去执行任务 因为已经把任务加入到阻塞队列中了 而工作线程为0 所以需要addWorker()添加工作线程执行任务
addWorker(null, false);
}
//表示加入队列失败 那么addworker 新增一个非核心工作线程
//返回false 表示线程池状态不是running 或 工作线程数量超过限制
else if (!addWorker(command, false))
//执行拒绝策略
reject(command);
}接下来分别针对addWork()方法和reject()方法进行分析
addWorker()
addWorker()方法会检验线程池的状态，如果不是运行状态，直接返回false，根据core标志判断工作线程是否达到线程限制（参数core为true时，不能超过核心线程数，为fasle时不能超过最大线程），没有达到限制的话创建一个worker对象，而worker对象是一个线程，进一步判断线程池状态，如果是运行状态，那么调用start()方法启动worker线程，启动之后肯定会执行Worker对象中的run()方法。而如果启动线程失败，会调用addWorkerFailed()方法。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
// Check if queue empty only if necessary.
//判断线程池状态 如果已经不是运行状态 那么直接返回false
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
for (;;) {
//校验当前线程数量
//如果core标识为true 那么如果大于核心线程数就返回fasle
//如果core标识为false,大于最大线程数返回false
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
//通过cas新增线程数量 新增成功跳出自旋
if (compareAndIncrementWorkerCount(c))
//跳出retry指定的for循环 在这里也就是跳出了外面的for循环了
break retry;
//重新获取c的值 因为如果上面cas失败了 那么c的值肯定是变动了 所以重新获取一下
c = ctl.get(); // Re-read ctl
//如果线程池的状态已经是shoudown 或 stop等非运行的状态 那么跳过这次的循环
if (runStateAtLeast(c, SHUTDOWN))
//重新进入到外面的for循环
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//工作线程是否启动的标识
boolean workerStarted = false;
//工作线程是否添加成功的标识
boolean workerAdded = false;
Worker w = null;
try {
// work实现了Runnable 是一个线程 这里只是创建一个线程 还没有开始运行
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//加锁
mainLock.lock();
try {
int c = ctl.get();
//判断线程池的状态
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
//当前的工作线程不是新生状态 抛出异常
if (t.getState() != Thread.State.NEW)
throw new IllegalThreadStateException();
//把woker对象放入workers中 这是一个HashSet
workers.add(w);
//修改工作线程是否添加成功的标识
workerAdded = true;
int s = workers.size();
if (s > largestPoolSize)
//更新largestPoolSize 用于统计线程池的最大size 用于监控数据
largestPoolSize = s;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//如果已经添加成功 那么启动工作线程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//启动失败 那么处理失败的情况
addWorkerFailed(w);
}
return workerStarted;
}addWorkerFailed()
启动线程失败时调用
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
//加锁
mainLock.lock();
try {
if (w != null)
//从set中移除启动失败的woker
workers.remove(w);
//将工作线程数减一 已经加锁 不需要通过cas来修改了
decrementWorkerCount();
//根据线程池的状态看是否需要停止线程池
tryTerminate();
} finally {
mainLock.unlock();
}
}可以看出worker类实现了Runnable接口，是一个线程，并且在构造函数中将任务赋值给firstTask成员变量，并且通过线程工厂将当前worker对象封装之后保存在thread成员变量中。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
final Thread thread;
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
}接下来分析worker对象中的run()方法，run()方法会调用runWorker()方法
public void run() {
runWorker(this);
}复制代码runWorker()
要使工作线程能够复用，所以run()方法不能结束，所以采用while()循环来保证有任务的情况下run()方法是不会结束的。firstTask为空的情况下，在调用getTask()获取到任务之后会执行任务的run()方法，注意不是执行start()方法，因为当前的worker对象已经是一个线程，所以直接调用任务的run()方法执行里面的逻辑就行了。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//这里是while循环 用来保证有任务的时候run方法不执行结束
//while这里的判断也就是说 task为空而且获取getTask()返回为空 那么线程就会结束
//注意 这里是短路或 说明如果woker对象中的firstTask不为空 就不会调用getTask()
//所以getTask()返回null 意味着结束当前woker线程
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
//判断线程池状态 如果线程池已经停止 那么任务线程中断
//如果不是 那么确保线程不中断
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//执行任务之前的处理 默认是空实现
//继承ThreadPoolExecutor 重写方法 可以加入自己的逻辑
beforeExecute(wt, task);
try {
//调用任务的run方法 注意这里是调用传入的任务的run方法
//因为已经创建了work线程来执行 所以不是调用start()方法执行任务
task.run();
//执行任务之后的处理 默认是空实现
//继承ThreadPoolExecutor 重写方法 可以加入自己的逻辑
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}接下来分析getTask()方法
getTask()
通过从阻塞队列中获取任务，如果当选线程数量大于核心线程那么通过poll(timeout,TimeUnit)来阻塞工作线程，否则直接调用take()方法一直阻塞，直到获取到任务，然后返回。（注意：从这里我们可以知道，所谓的核心线程和非核心线程不是线程的属性，线程池只要维护核心线程的数量即可）
private Runnable getTask() {
// 用来表示上一次循环的时候，非核心线程获取任务是否超时
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
// Check if queue empty only if necessary.
// 如果线程池已经不是running状态 而且队列是空的
//那么将工作线程数量减1并返回null 也就是结束当前work线程
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//获取工作线程数量
int wc = workerCountOf(c);
// Are workers subject to culling?
//allowCoreThreadTimeOut 默认为false 标识核心线程在空闲时也不回收
//所以这里也就是工作线程数大于核心线程数的时候 timed为true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//这个判断标识如果工作线程数已经大于最大线程数并且大于1 而且队列已经是空
//(如果上一次循环的时候，非核心线程获取任务超时，那么timedOut为true)
//那么将cas将工作线程数减1 并返回null 结束当前work线程
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
//如果cas失败 那么证明工作线程数已经变化 重新进入for循环
continue;
}
try {
//工作线程数大于核心线程数的时候 timed为true
//从阻塞队列中获取任务 如果工作线程数大于核心线程数 那么利用配置的最大空闲时间和时间单位对线程进行有时间限制的阻塞 否则调用take阻塞
//从这里我们可以知道工作线程数小于核心线程数是不会回收的 一直阻塞在这
//而超过核心线程数 那么需要根据配置空闲时间来回收线程
//poll(timeout,unit)超时没有获取到会返回null，而take()方法会一直阻塞在这 知道获取到阻塞队列中的任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
//获取到任务直接返回 从这里可以知道非核心线程也不是超时了就结束掉（并没有返回null,），而是进入下一次循环 还有一次获取任务的机会
if (r != null)
return r;
//否则将timedOut设置为true 进入下一次循环
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}addWorker()的部分分析完之后，接下来分析reject()方法
reject()
reject()方法很简单，根据构造函数传过来的拒绝策略调用相应的rejectedExecution()方法（策略模式）。
final void reject(Runnable command) {
//根据拒绝策略的参数分别执行相应的操作
handler.rejectedExecution(command, this);
}// 1. 当前线程执行任务
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
//2.默认 直接抛出异常
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());
}
}
//3.什么都不做
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
//4.丢弃队列头部的任务，也就是最旧的任务，并将当前任务调用execute()
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}七：结束语
线程池在我们工作中，使用的场景还是比较多的，而且线程是比较重要的资源，所以理解线程池的原理，合理利用线程池是很有必要的。
腾讯云开发者
