Java并发编程系列34 | 深入理解线程池（下）
2021-03-15 01:32
标签：throwable keep stop 详细 zookeeper final 话题 第一个 des 收录于话题 2020年Java原创面试题库连载中 本文是深入理解线程池下篇： 这一节详细分析任务提交到线程池之后，线程池是如何处理和执行任务的。 execute()方法执行过程如下： addWorkerr(Runnable firstTask, boolean core)方法的主要工作是在线程池中创建一个新的线程并执行： 线程池中的每一个线程被封装成一个Worker对象，线程池维护的其实就是一组Worker对象。 循环从等待队列中获取任务并执行： 获取workQueue中的任务 关闭线程池使用shutdown方法或shutdownNow方法，最终目的是将线程池状态设置成TERMINATED。 shutdown方法过程： 结束线程池，最终将线程池状态设置为TERMINATED。 shutdownNow方法过程： shutdown方法 VS shutdownNow方法 一些不需要长篇大论介绍的知识点，这里简单说下。 构造线程时传入的 RejectedExecutionHandler 类型参数 handler 就是拒绝策略。 ThreadPoolExecutor 中有四个已经定义好的RejectedExecutionHandler实现类可供我们直接使用。（我们也可以实现自己的策略） 我们可以通过线程池提供的参数和方法对线程池进行监控： 默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。如果需要线程池创建之后立即创建线程，可以通过以下两个方法实现： ThreadPoolExecutor提供了动态调整线程池容量大小的方法： 【原创】01|开篇获奖感言 之前，给大家发过三份Java面试宝典，这次新增了一份，目前总共是四份面试宝典，相信在跳槽前一个月按照面试宝典准备准备，基本没大问题。 看到这里，证明有所收获 Java并发编程系列34 | 深入理解线程池（下） 标签：throwable keep stop 详细 zookeeper final 话题 第一个 des 原文地址：https://blog.51cto.com/15009303/2552572
#java 976 #程序员 2286 #并发编程 238 #进阶架构师 | 并发编程专题 12
★★★建议星标我们★★★
公众号改版后文章乱序推荐，希望你可以点击上方“Java进阶架构师”，点击右上角，将我们设为★“星标”！这样才不会错过每日进阶架构文章呀。
【000期】Java最全面试题库思维导图
【001期】JavaSE面试题（一）：面向对象
【002期】JavaSE面试题（二）：基本数据类型与访问修饰符
【003期】JavaSE面试题（三）：JavaSE语法（1）
【004期】JavaSE面试题（四）：JavaSE语法（3）
【005期】JavaSE面试题（五）：String类
【006期】JavaSE面试题（六）：泛型
【007期】JavaSE面试题（七）：异常
【008期】JavaSE面试题（八）：集合之List
【009期】JavaSE面试题（九）：集合之Set
【010期】JavaSE面试题（十）：集合之Map
【011期】JavaSE面试题（十一）：多线程（1）
【012期】JavaSE面试题（十二）：多线程（2）
【013期】JavaSE面试题（十三）：多线程（3）
【014期】JavaSE面试题（十四）：基本IO流
【015期】JavaSE面试题（十五）：网络IO流
【016期】JavaSE面试题（十六）：反射
【017期】JavaSE面试题（十七）：JVM之内存模型
【018期】JavaSE面试题（十八）：JVM之垃圾回收
【020期】JavaSE系列面试题汇总（共18篇）
【019期】JavaWeb面试题（一）：JDBC
【021期】JavaWeb面试题（二）：HTTP协议
【022期】JavaWeb面试题（三）：Cookie和Session
【023期】JavaWeb面试题（四）：JSP
【024期】JavaWeb面试题（五）：Filter和Listener
【025期】Java工具面试题（一）：版本控制工具
【026期】Java工具面试题（二）：项目管理工具
【027期】Java设计模式面试题
【028期】JavaWeb系列面试题汇总（共10篇）
【029期】JavaEE面试题（一）Web应用服务器
【030期】JavaEE面试题（二）SpringMVC
【031期】JavaEE面试题（三）Spring（1）
【032期】JavaEE面试题（四）Spring（2）
【033期】JaveEE面试题（五）MyBatis
【034期】JavaEE面试题（六）Hibernate
【035期】JavaEE面试题（七）SpringBoot（1）
更多内容，点击上面蓝字查看
线程池介绍
Executor框架接口
线程池状态
线程池参数
线程池创建
执行过程
关闭线程池
其他问题
任务拒绝策略
线程池中的线程初始化
线程池容量的动态调整
线程池的监控6. 执行过程
6.1 execute()方法
1.如果workerCount 2.如果workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中；
3.如果workerCount >= corePoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务；
4.如果workerCount >= maximumPoolSize，且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
/*
* 当前核心线程数小于corePoolSize，则新建一个线程放入线程池中。
* 注意这里不管核心线程有没有空闲，都会创建线程
*/
if (workerCountOf(c) 6.2 addWorker方法
参数firstTask：这个新创建的线程需要第一个执行的任务；firstTask==null，表示创建线程，到workQueue中取任务执行；
参数core：true代表使用corePoolSize作为创建线程的界限；false代表使用maximumPoolSize作为界限
/**
* @return
*/
private boolean addWorker(Runnable firstTask, boolean core) {
// 1. 增加线程数量ctl
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);// 获取运行状态
/*
* 不能创建线程的几种情况：
* 1. 线程池已关闭且rs == SHUTDOWN，不允许提交任务，且中断正在执行的任务
* 2. 线程池已关闭且firstTask!=null，
* 3. 线程池已关闭且workQueue为空
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);// 获取线程数
// 判断线程数上限
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 尝试增加workerCount，如果成功，则跳出外层for循环
if (compareAndIncrementWorkerCount(c))
break retry;
// CAS失败，循环尝试
c = ctl.get();
if (runStateOf(c) != rs)
continue retry;
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
/*
* 2. 创建Worker对象来执行任务，每一个Worker对象都会创建一个线程
* Worker类下文详细讲解
*/
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
/*
* 判断状态：
* 小于 SHUTTDOWN 那就是 RUNNING，最正常的情况
* 等于 SHUTDOWN，不接受新的任务但是会继续执行等待队列中的任务，所以要求firstTask == null
*/
if (rs largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 3. worker添加成功，启动这个worker中的线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}6.3 Worker类
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
// 线程被封装成Worker
final Thread thread;
/*
* 在创建线程的时候，如果同时指定的需要执行的第一个任务。
* 可以为 null，线程自己到任务队列中取任务执行
*/
Runnable firstTask;
// 线程完成的任务数
volatile long completedTasks;
// Worker 只有这一个构造方法
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);// 调用 ThreadFactory 来创建一个新的线程
}
/**
* worker工作，调用外部类的 runWorker 方法，循环等待队列中获取任务并执行，下文详细介绍
*/
public void run() {
runWorker(this);
}
// ... 其他几个方法用AQS及锁的操作，不关注
}6.4 runWorker方法
/**
* 循环从等待队列中获取任务并执行
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
/*
* 循环调用 getTask() 获取任务，getTask()下文详细讲解
* 获取到任务就执行，
* 获取不到就阻塞等待新任务，
* 返回null任务就销毁当前线程
*/
while (task != null || (task = getTask()) != null) {
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();// 真正执行任务，执行execute()中传入任务的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 {
// 如果到这里，需要销毁线程：
// 1. getTask 返回 null退出while循环，队列中没有任务或空闲线程超时
// 2. 任务执行过程中发生了异常
processWorkerExit(w, completedAbruptly);
}
}6.5 getTask方法
/**
* 获取workQueue中的任务
* 1. 正常情况，直接workQueue.take()获取到任务返回；
* 2. workQueue中没有任务，当前线程阻塞直达获取到任务；
* 3. getTask()返回 null， runWorker()方法会销毁当前线程，如下情况返回null：
* 状态为SHUTDOWN && workQueue.isEmpty()
* 状态 >= STOP
* 当前线程数 wc > maximumPoolSize
* 空闲线程超时keepAliveTime
*/
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
/*
* 两种返回null的情况：
* 1. rs == SHUTDOWN && workQueue.isEmpty()
* 2. rs >= STOP
*/
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();// CAS 操作，减少工作线程数
return null;
}
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/*
* 两种返回null的情况：
* 1. 当前线程数 wc > maximumPoolSize，return null
* 2. 空闲线程超时，return null
*/
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
/*
* 到 workQueue 中获取任务并返回
*/
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}6.6 总结
7. 关闭线程池
7.1 shutdown方法
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();// 安全策略判断
advanceRunState(SHUTDOWN);// CAS设置线程池状态为SHUTDOWN
interruptIdleWorkers();// 中断空闲线程
onShutdown(); // 钩子方法，用于ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 尝试结束线程池，下文详细讲解
tryTerminate();
}7.2 tryTerminate方法
/**
* 结束线程池，最终将线程池状态设置为TERMINATED
*/
final void tryTerminate() {
for (;;) {
int c = ctl.get();
/*
* 当前线程池的状态为以下几种情况时，直接返回：
* 1. RUNNING，因为还在运行中，不能停止；
* 2. TIDYING或TERMINATED，已经关闭了；
* 3. SHUTDOWN并且等待队列非空，这时要执行完workQueue中的task；
*/
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 如果线程数量不为0，则中断一个空闲的工作线程，并返回
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 尝试设置状态为TIDYING
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();// 钩子方法，留给子类实现
} finally {
// 设置状态为TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}7.3 shutdownNow方法
public List
shutdown方法设置线程池状态为SHUTDOWN，SHUTDOWN状态不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务。
shutdownNow方法设置线程池状态为STOP，STOP状态不能接受新任务，也不处理队列中的任务，会中断正在处理任务的线程。8. 其他问题
8.1 任务拒绝策略
RejectedExecutionHandler只有一个钩子方法，执行拒绝策略。
public interface RejectedExecutionHandler {
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
/**
* 由提交任务的线程自己来执行这个任务
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
/**
* 默认的策略：接抛出 RejectedExecutionException 异常
*/
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());
}
}
/**
* 不做任何处理，直接忽略掉这个任务
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
/**
* 把队列队头的任务(也就是等待了最长时间的)直接扔掉，然后提交这个任务到等待队列中
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}8.2 线程池的监控
8.3 线程池中的线程初始化
prestartCoreThread()：初始化一个核心线程；
prestartAllCoreThreads()：初始化所有核心线程。8.4 线程池容量的动态调整
并发系列文章汇总
【原创】02|并发编程三大核心问题
【原创】03|重排序-可见性和有序性问题根源
【原创】04|Java 内存模型详解
【原创】05|深入理解 volatile
【原创】06|你不知道的 final
【原创】07|synchronized 原理
【原创】08|synchronized 锁优化
【原创】09|基础干货
【原创】10|线程状态
【原创】11|线程调度
【原创】12|揭秘 CAS
【原创】13|LockSupport
【原创】14|AQS 源码分析
【原创】15|重入锁 ReentrantLock
【原创】16|公平锁与非公平锁
【原创】17|读写锁八讲（上）
【原创】18|读写锁八讲（下）
【原创】19|JDK8新增锁StampedLock
【原创】20|StampedLock源码解析
【原创】21|Condition-Lock的等待通知
【原创】22|倒计时器CountDownLatch
【原创】22|倒计时器CountDownLatch
【原创】23|循环屏障CyclicBarrier
【原创】24|信号量Semaphore
【原创】25|交换器Exchangere
【原创】26|ConcurrentHashMap（上）
【原创】27|ConcurrentHashMap（下）
【原创】28|Copy-On-Write容器
【原创】29|ConcurrentLinkedQueue
【原创】30 | ThreadLocal
【原创】31 | 阻塞队列（上）
《java面试宝典5.0》(初中级)
《350道Java面试题：整理自100+公司》（中高级）
《资深java面试宝典-视频版》（资深）
《Java[BAT]面试必备》（资深）
分别适用于初中级，中高级，资深级工程师的面试复习。
内容包含java基础、javaweb、mysql性能优化、JVM、锁、百万并发、消息队列，高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper、数据结构、限流熔断降级等等。
获取方式：点“在看”，V信关注上述Java最全面试题库号并回复 【面试】即可领取，更多精彩陆续奉上。
必须点个在看支持呀，喵