一、线程池基本概念和线程池前置知识
1.1 Java中创建线程的方式有哪些
传统答案：
- 继承Thread类
通过继承Thread类并重写其run方法来创建线程。具体步骤包括定义Thread类的子类，在子类中重写run方法以实现线程的具体逻辑，然后创建子类的实例并调用其start方法来启动线程。
- 实现Runnable接口
通过实现Runnable接口并重写其run方法来创建线程。这种方式相较于继承Thread类更为灵活，因为Java不支持多重继承，而实现接口则没有这个问题。实现Runnable接口后，需要将其实现类的实例作为参数传递给Thread类的构造函数，然后调用Thread对象的start方法来启动线程。
- 实现Callable接口
Callable接口与Runnable接口类似，但其call方法可以返回值，并且可以抛出异常。创建Callable实现类的实例后，通常使用FutureTask类来包装它，并将FutureTask对象作为Thread对象的target来创建并启动新线程。调用FutureTask对象的get方法可以获取子线程执行结束后的返回值。
- 创建线程池
Executor框架提供了多种线程池的实现，如单线程池（newSingleThreadExecutor）、固定大小的线程池（newFixedThreadPool）、可缓存的线程池（newCachedThreadPool）以及定时线程池（newScheduledThreadPool）等。使用线程池可以更有效地管理线程的创建、销毁和调度，从而提高系统的性能和稳定性。
1.2 为什么要使用线程池
- 降低资源消耗：在创建和销毁线程时，会消耗一定的系统资源，包括CPU和内存。通过线程池，可以重复利用已创建的线程，从而降低这种消耗。
- 提高响应速度：当新任务到达时，如果线程池中已有空闲线程，则任务可以立即执行，无需等待线程的创建。
- 提高线程的可管理性：线程是稀缺资源，如果无限制地创建线程，不仅会消耗大量系统资源，还可能降低系统的稳定性和性能。线程池提供了统一的线程管理和调优机制，可以方便地对线程进行监控和管理。
1.3 线程池的核心参数
在源码中找到ThreadPoolExecutor类中参数最多的一个构造方法
这七个就是线程池的核心参数：
- int corePoolSize,
- int maximumPoolSize,
- long keepAliveTime,
- TimeUnit unit,
- BlockingQueue<Runnable> workQueue,
- ThreadFactory threadFactory,
- RejectedExecutionHandler handler
- corePoolSize（核心线程数）：
- 含义：线程池中始终保持活动的线程数量，即使这些线程处于空闲状态，也不会被销毁(默认情况下)。
- 作用：当任务提交到线程池时，如果当前线程数未达到corePoolSize，则会创建新线程来执行任务。
- maximumPoolSize（最大线程数）：
- 含义：线程池中允许的最大线程数量。
- 作用：当任务队列已满，且当前线程数小于maximumPoolSize时，线程池会创建新的线程来处理任务。但线程池不会无限制地创建线程，最大线程数由maximumPoolSize指定。
- keepAliveTime（空闲线程存活时间）：
- 含义：当线程池中的线程数量超过corePoolSize时，空闲线程在指定的时间内没有被使用时，将被销毁，直到线程池中的线程数量减少到corePoolSize为止。
- 计量单位：由
unit参数指定。
- unit（时间单位）：
- 含义：keepAliveTime的时间单位。
- 可选值：TimeUnit.DAYS（天）、TimeUnit.HOURS（小时）、TimeUnit.MINUTES（分）、TimeUnit.SECONDS（秒）、TimeUnit.MILLISECONDS（毫秒）、TimeUnit.MICROSECONDS（微秒）、TimeUnit.NANOSECONDS（纳秒）。
- workQueue（工作队列）：
- 含义：用于保存等待执行的任务的队列。
- 类型：线程池提供了多种工作队列的实现，如ArrayBlockingQueue（基于数组的有界阻塞队列）、LinkedBlockingQueue（基于链表的无界阻塞队列）、SynchronousQueue（一个不缓存任务的阻塞队列）等。
- 作用：当新任务提交到线程池时，如果线程数未达到corePoolSize，则创建新线程处理；如果线程数已达到corePoolSize但小于maximumPoolSize，且工作队列未满，则将任务放入工作队列；如果工作队列已满且线程数小于maximumPoolSize，则创建新线程处理；如果工作队列已满且线程数已达到maximumPoolSize，则执行拒绝策略。
- threadFactory（线程工厂）：
- 含义：用于创建新线程的工厂。
- 作用：可以自定义线程的名称、优先级等属性。
- handler（拒绝策略）：
- 含义：当任务无法加入工作队列且线程池已达到最大线程数时，用于处理新提交的任务的策略。
- 类型：常见的拒绝策略包括抛出异常、丢弃任务、丢弃队列中最老的任务、将任务分给调用线程来执行等。
二、线程池任务处理策略
线程池执行任务的方法是execute方法。
想要查看执行流程的话，需要查看的就是execute方法的源码。
将源码文本粘贴出来，逐步分析：
// 任务交给线程池处理时，一般会执行execute方法，并传递任务
// command 就是传递过来的任务
public void execute(Runnable command) {
// 非空校验
if (command == null)
throw new NullPointerException();
// 以下是核心业务流程
// ctl 是什么？ctl是线程池的一个核心属性。
// 想要了解线程池的执行流程需要先知道线程池的核心属性
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
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);
}
else if (!addWorker(command, false))
reject(command);
}由于在execute方法中使用到了线程池的核心属性 ctl，所以我们先看一下ctl
粘贴出来分析：
// 线程池的核心属性
// AtomicInteger是系统底层保护的int类型，通过对int类型的数据进行封装，提供执行方法的控制进行值的原子操作。
// 可以理解为private final int ctl = 0;
// ctl 存储了线程池的两个核心属性：线程池状态和工作线程个数
// int类型占32个比特位
// 线程池状态：基于ctl的高三位存储线程池状态
// 工作线程个数：基于ctl的低29位存储工作线程个数
// 那么线程池中最多可以有多少个工作线程呢？答案是2^29个
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// Integer.SIZE=32 所以COUNT_BITS =29
private static final int COUNT_BITS = Integer.SIZE - 3;
// CAPACITY 就是2^29 也就是线程池中工作线程数的最大值
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 下面5个属性是线程池的状态
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;线程池的生命周期（五个状态）：
整明白了线程池的核心属性，下面可以继续看ThreadPoolExecutor的execute方法了
// 任务交给线程池处理时，一般会执行execute方法，并传递任务
// command 就是传递过来的任务
public void execute(Runnable command) {
// 非空校验
if (command == null)
throw new NullPointerException();
// 以下是核心业务流程
// ctl.get()：拿到存储线程池状态和工作线程个数的核心属性
int c = ctl.get();
// workerCountOf（） 获取工作线程个数 corePoolSize：核心线程数
// 判断当前工作线程数是否小于核心线程数（构建线程池时指定的）
if (workerCountOf(c) < corePoolSize) {
// 创建工作线程
//第一个参数 command ：传递的任务
//第二个参数 是否是核心线程，true：创建核心线程；false：创建非核心线程
// addWorker方法 返回值是布尔类型 代表创建是否成功
if (addWorker(command, true))
// 创建成功，结束，任务给核心线程处理
return;
//创建失败 重新获取核心属性ctl
c = ctl.get();
}
// 如果当前工作线程数已经达到核心线程数 执行下面的语句
// isRunning(c) 判断当前线程池是不是running状态 如果是直接将任务扔进工作队列
// 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)
addWorker(null, false);
}
// 尝试构建非核心线程去处理当前任务
else if (!addWorker(command, false))
// 如果非核心线程创建失败 执行拒绝策略
reject(command);
}从上面源码可以看出，执行流程为：核心线程-工作队列-非核心线程-拒绝策略
如果想要修改默认的顺序可以重写execute方法来实现。
问题1：
下面代码块，是在工作线程数量小于核心线程数量时执行的代码，作用是，工作线程数量小于核心线程数量时，创建一个核心线程，并把任务给这个核心线程。
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
// 创建成功，结束，任务给核心线程处理
return;
//创建失败 重新获取核心属性ctl
c = ctl.get();
}
那么问题是为什么创建核心线程失败后要使用 c = ctl.get();来重新获取核心属性呢？
举个例子来回答这个问题，假如核心线程数是5，当前工作线程数是4，在并发情况下，有两个线程同时进入到if代码块里面，都去执行addWorker(command, true)方法，创建核心工作线程，由于addWorker方法内部通过一定的方式保证了原子性，所以只能创建成功1个核心工作线程，另一个不会创建成功（返回false），这种情况，不会直接return，会去继续执行方法下面的代码，下面的代码会使用到核心线程属性ctl，而此时显然ctl核心线程属性已经发生了变化（另外一个线程创建成功了一个工作线程，工作线程数量发生了变化），需要重新获取最新的。
问题2：
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);
}这段代码中，主要实现的是如果线程池是Running状态，将任务放入工作队列的功能。
那么isRunning(c) 和workQueue.offer(command)两句代码不是就能实现这个功能了吗，为什么还要写上述代码？
因为在多线程环境中，当执行完isRunning(c) 和执行workQueue.offer(command)之间的间隙，线程池的状态可以就发生了变化。
比如一开始线程池是Running状态，执行isRunning(c)得到返回值true，然后在执行workQueue.offer(command)代码前线程池状态改变，不再是Running状态了，不过由于之前获取到的isRunning(c)的返回值是true，所以还是会执行workQueue.offer(command)方法。
这样显然是有问题的，所以为了应对这种情况，在下面添加了以下代码：
// ctl.get();重新获取线程池核心属性
int recheck = ctl.get();
// 根据核心属性 重新校验线程池是否是Running状态，如果不是Running状态，移除前边添加的任务并执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
// 如果线程池状态还是为Running状态 则执行下面语句
// 如果线程池仍然在运行，但此时工作线程的数量为0（workerCountOf(recheck)返回0），则调用addWorker(null, false)方法来添加一个新的核心线程。
//null 作为第一个参数，表示这个新线程在启动时不带有任何初始任务 ，false代表创建非核心线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
关于 addWorker(null, false)：
-
firstTask参数允许你指定一个新创建的线程应该首先运行的任务。如果firstTask是null，那么新线程将不会立即执行任何任务，而是会等待从工作队列中取出一个任务来执行。 -
在线程池的工作机制中，如果当前线程数少于
corePoolSize（核心线程数），并且有新任务提交，那么线程池会尝试启动一个新线程来执行这个任务，而不是将任务放入工作队列。
三、创建工作线程的流程
查看创建工作线程的流程主要是查看addWorker方法的源码
对源码逐步解析:
// 创建工作线程（核心线程和非核心线程都是基于addWorker创建的）
// 第一个参数：任务，第二个参数：指定是核心还是非核心
private boolean addWorker(Runnable firstTask, boolean core) {
// 第一部分代码
// 做两件事情：
// 1.判断线程池状态（外层for循环）
// 2.判断线程个数（内层for循环），然后基于cas修改ctl属性，给工作线程个数+1
retry: // 给外层for循环取一个名称retry
for (;;) { // 死循环 相当于while（true）
int c = ctl.get(); // 拿到核心属性ctl
int rs = runStateOf(c); // rs:高三位的线程池状态
// rs >=SHUTDOWN 即rs为SHUTDOWN/STOP/TIDYING/TERMINATED 即不是Running状态
// 根据前面线程池五个属性的那张图可知线程池不能接收新任务
if (rs >= SHUTDOWN &&
// 第二个判断 解决在SHUTDOWN 状态下，没有工作线程，但是工作队列有任务
// 要构建一个线程处理阻塞队列（工作队列）任务的情况
// rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty())的情况不能走return语句
! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c); // 工作线程个数
// CAPACITY :工作线程最大值
if (wc >= CAPACITY ||
// 核心线程：判断corePoolSize
// 非核心线程：判断maximumPoolSize
// core 是addWorker的第二个参数 true代表核心线程，false代表创建非核心线程
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 为什么要用CAS的方式修改？
// 为了避免多线程并发创建工作线程，导致破坏设置的核心参数（比如设置核心参数是5，创建了6个）
if (compareAndIncrementWorkerCount(c))
// 如果成功 跳出外层循环
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// 第二部分代码
// 做两件事情：1.创建工作线程 2.启动工作线程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;// Worker 对象就是工作线程
try {
// 创建工作线程，并把任务交给Worker 对象
w = new Worker(firstTask);
// 将new Worker时创建的thread拿到
final Thread t = w.thread;
// 判断使用线程池的用户，指定的线程工厂构建的thread不是null
if (t != null) {
// 同步锁 为什么要加锁？
// 下面workers.add(w); 按住Ctrl+单击 查看可知 workers是HashSet类型
// HashSet不是线程安全的 为了保证线程安全，所以加锁了，不加锁不安全
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 如果工作线程添加成功 则启动该线程
if (workerAdded) {
t.start();// 启动一个线程，执行run方法
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}根据源码总结：
线程池在创建工作线程时：
- 判断线程池是否符合要求。
- 判断工作线程个数是否符合要求，并且基于CAS保证原子性。
- new Worker创建工作线程，并添加HashSet中，基于ReentrantLock保证原子性。
- 拿到Worker对象中的thread，执行start方法，启动线程。
四、工作线程Worker
线程池中的工作线程就是Worker对象，查看一下Worker里面做了什么事情。
// Worker是工作线程，Worker也会存储一个任务（只存储第一个任务）
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 当前对象
// 创建的线程调用start启动时，执行的是谁的run方法？
// 执行的是Worker对象里面的run方法
this.thread = getThreadFactory().newThread(this);
}
// addWorker方法中执行thread.start()方法后，执行的是Worker对象中的run方法
public void run() {
runWorker(this);
}
}AQS是什么?
AQS是AbstractQueuedSynchronizer，是JUC包下的并发基础类，很多同步内容都是基于AQS实现的，比如：
- ReentrantLock
- ReentrantReadWriteLock
- CountDownLatch
- Semaphore
- 线程池中的Worker对象也是基于AQS做了一个实现
Worker继承AQS干嘛？
Worker线程继承了AQS后，可以使用基于CAS修改的属性state
在shutdown状态下空闲线程要执行interrupt中断，工作中的线程，不能执行interrupt
工作线程在处理任务前，会先执行lock方法（将state从0改为1），也就是正在干活的线程state==1
在终端线程前，判断每一个线程的state，如果为0直接interrupt，如果为1什么也不做。
五、执行/拉取任务流程
5.1 执行任务
执行任务，就是启动工作线程后，执行了Worker对象中的run方法，run方法中执行了runWorker方法。
工作线程直接通过task.run执行任务，并且线程池预留了beforeExecute和afterExecute方法，可以在任务执行前后做一些额外处理。
// runWorker传递的参数就是Worker对象本身
final void runWorker(Worker w) {
// 拿到工作线程中的thread
Thread wt = Thread.currentThread();
// 拿到Worker对象中存储的第一个任务
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 如果Worker对象在启动时携带了任务，那就优先执行携带的任务
while (task != null || (task = getTask()) != null) {
w.lock();
// 判断线程池状态是不是stop 如果时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);
}
}5.2 拉取任务
工作线程在处理完自带的任务后会直接基于getTask方法，从阻塞队列中拉取任务。
如果是核心线程，默认情况下，会基于take方法在工作队列中拉取任务。
如果是非核心线程，会基于poll方法，拉取指定时间任务。（时间到了直接告辞）
// 工作线程从阻塞队列拉取任务的操作
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 判断线程池状态 如果状态已经变为stop或者状态为shutdown且工作队列任务都处理完毕
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
// 工作线程个数-1 并且干掉当前工作线程
// 线程正常死亡：run方法结束
decrementWorkerCount();
return null;
// return null： runWarker方法中 while循环结束，线程正常消亡；
// 因为是Worker类中的run 方法中调用的runWarker方法，并且run方法中没有其他代码
// 所以runWarker结束，run方法也就结束了，线程也就消亡了
}
int wc = workerCountOf(c);
// 核心线程执行take 非核心线程执行poll方法（poll方法拉取最大空闲时间）
// 线程池中的核心线程可以基于keepAliveTime（最大空闲时间）去结束吗？
// 或者说线程池中的核心线程一定会永远存放在线程池里面吗？
// 不一定 有一个属性allowCoreThreadTimeOut：是否允许核心线程超时，默认是false，但可以设置为true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
// poll方法，拉取阻塞队列任务，指定keepAliveTime时间
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
// take方法，死等任务，知道中断
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}六、线程池关闭流程和锁
shutdown方法：
- 将线程池状态修改为shutdown
- 将空闲的工作线程直接中断
- 在确认工作队列中的任务全部处理完并且工作线程个数为0，自动改为tidying状态
- 在tidying状态下，执行terminated方法，变为terminated状态
- 到这线程池结束
// 优雅的关闭线程池
// 为什么要加锁？
// 会操作Worker对象
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);// 线程池状态修改为shutdown
interruptIdleWorkers();// 中断空闲的工作线程
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}shutdownNow方法：
- 将线程池状态修改为stop
- 将所有的工作线程直接中断
- 在工作线程个数为0之后，自动改为tidying状态
- 在tidying状态下，执行terminated方法，变为terminated状态
- 到这线程池结束
在工作中核心线程数、最大线程数、最大空闲时间、任务队列怎么设置比较好？
如果为了充分发挥硬件性能，一般只需要考虑三个信息的设置
- 核心线程数：最重要的属性，你要根据你的任务类型，判断线程数设置多少合适
任务类型：
CPU密集型任务：线程一直在干活，不希望CPU做上下文切换。
io密集型任务：因为线程干一会儿歇一会儿。
混合型任务：因为混合型偶尔要求CPU一直调度，偶尔不干活，可以切换。
想要设置好核心线程数，去发挥服务器硬件性能，需要动态的调试和压测。为了避免调试参数时反复重启，并且成本太高，可以直接设置动态线程池，因为线程池提供了Set方法设置核心参数，以及get方法查看核心参数，可以在压测时，根据CPU占用率和使用情况，来调整核心线程数。
- 工作队列长度：根据服务器内存使用情况来调试，同时也要查看好jvm的堆内存大小。
- 拒绝策略：根据任务来决定，如果任务就是写个日志啥的，那就直接扔了，如果任务是必须要执行的，那就直接重试，或者存储到数据库，后期做同意补偿操作。
![[Linux] TCP协议介绍(3): TCP协议的“四次挥手“过程、状态分析...](https://img-blog.csdnimg.cn/img_convert/40f4ab113adb2f5329ba24ca8005239c.webp?x-oss-process=image/format,png)
