Java 入门指南：Java 并发编程 —

线程池

线程池（ThreadPool）是一种用于管理和复用线程的机制，它可以预先创建一批线程，并维护一个线程队列，用于执行提交的任务。

线程池的主要目的是提高多线程应用程序的性能和效率，通过重用已创建的线程，减少线程的创建和销毁开销，避免频繁地创建线程和线程上下文切换的性能损耗。

线程池其实是一种池化的技术实现，池化技术的核心思想就是实现资源的复用，避免资源的重复创建和销毁带来的性能开销。

线程池可以管理一系列线程，让线程执行完任务之后不进行销毁，而是继续去处理其它线程已经提交的任务。

应用场景

在 Java 程序中，其实经常需要用到多线程来处理一些业务，但是不建议单纯继承 Thread 类或者实现 Runnable 接口来创建线程，这样会导致频繁创建及销毁线程，同时创建过多的线程也可能引发资源耗尽的风险。

所以使用线程池是一种更合理的选择，方便管理任务，同时实现线程的重复利用。所以线程池一般适合需要异步或者多线程处理任务的场景。例如 Web服务器、并行计算、异步任务处理等场景。

线程池的优点

提高系统性能：线程池能够更好地利用系统资源，通过复用线程和减少线程的创建和销毁开销，有效提高系统的处理能力。
提高响应速度：通过线程池，可以将任务在多个线程上并发执行，当任务到达时，任务可以不需要等到线程创建就能立即执行，提高任务的响应速度。
提供线程管理和监控：线程池可以提供可配置的线程数量、线程保活时间、线程拒绝策略等功能，方便进行线程管理和监控。
避免资源耗尽：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性。通过限制线程数量，可以避免无限制地创建线程导致系统资源耗尽的风险。

使用线程池时，需要注意合理配置线程池的大小和参数，以及根据任务类型和性能需求选择合适的线程池类型。同时，需要注意处理任务执行过程中可能出现的异常和错误，以保证程序的稳定性。

线程池的基本概念

核心线程数（core pool size）：线程池中保持的线程数量，即使这些线程处于空闲状态，也至少会保留这么多线程。
最大线程数（maximum pool size）：线程池允许的最大线程数量，超过这个数量的请求将会排队或被拒绝。
工作队列（work queue）：用来存放等待执行的任务的队列。当线程池中的线程数量达到最大值时，新提交的任务将会被放入这个队列中等待执行。
拒绝策略（rejection policy）：当线程池无法处理更多任务时（即所有线程都在执行任务，并且工作队列已满），线程池会采用某种策略来处理这些任务。常见的拒绝策略包括丢弃任务、使用调用者所在的线程来执行任务等。
线程工厂（thread factory）：用于创建新线程的工厂类，可以根据需要定制新线程的属性。

线程池的构造

Java 提供了一个内置的线程池实现，即 java.util.concurrent.Executors 类。通过 Executors 类可以创建不同类型的线程池，例如固定大小线程池、缓存线程池、定时线程池等。

在这里插入图片描述

corePoolSize：线程池中用来工作的核心线程数量。
maximumPoolSize：最大线程数，线程池允许创建的最大线程数。
keepAliveTime：超出 corePoolSize 后创建的线程存活时间或者是所有线程最大存活时间，取决于配置。
unit：keepAliveTime 的时间单位。
workQueue：任务队列，是一个阻塞队列，当线程数达到核心线程数后，会将任务存储在阻塞队列中。
threadFactory ：线程池内部创建线程所用的工厂。
handler：拒绝策略；当队列已满并且线程数量达到最大线程数量时，会调用该方法处理任务。

创建线程池示例

下面是一个使用 ThreadPoolExecutor 创建线程池的示例：

import java.util.concurrent.*;public class ThreadPoolExample {public static void main(String[] args) {// 创建一个线程池ThreadPoolExecutor executor = new ThreadPoolExecutor(2, // 核心线程数5, // 最大线程数60, // 非核心线程的空闲超时时间（秒）TimeUnit.SECONDS, // 时间单位new ArrayBlockingQueue<>(10), // 工作队列Executors.defaultThreadFactory(), // 线程工厂new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略);// 提交任务给线程池for (int i = 0; i < 10; i++) {int taskId = i;executor.execute(() -> {System.out.println("正在执行任务 " + taskId + "，线程名称：" + Thread.currentThread().getName());try {Thread.sleep(1000); // 模拟任务执行时间} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}// 关闭线程池executor.shutdown();try {if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {executor.shutdownNow();}} catch (InterruptedException e) {executor.shutdownNow();Thread.currentThread().interrupt();}}
}

示例说明：

创建线程池：创建一个 ThreadPoolExecutor 实例，设置核心线程数为 2，最大线程数为 5，非核心线程的空闲超时时间为 60 秒，工作队列为 ArrayBlockingQueue，容量为 10，线程工厂使用默认的工厂，拒绝策略为 CallerRunsPolicy。
提交任务给线程池：提交 10 个任务给线程池，每个任务在执行时输出任务 ID 和当前线程的名称，并模拟任务执行时间为 1 秒。
关闭线程池，等待所有任务完成。如果在 60 秒内未完成，则强制关闭线程池。如果在等待过程中发生中断，则也强制关闭线程池并设置当前线程的中断状态。

ThreadFactory

ThreadFactory 接口是用来创建新的线程的工厂类。它提供了一个创建线程的方法 newThread(Runnable r)，根据传入的 Runnable 实例创建一个新的线程。

以下是 ThreadFactory 接口的代码示例：

public interface ThreadFactory {Thread newThread(Runnable r);
}

在使用线程池时，可以自定义实现 ThreadFactory 接口来创建线程，并通过 ThreadPoolExecutor 的构造方法将自定义的 ThreadFactory 对象传递给线程池。这样可以自定义线程的名称、优先级、是否为守护线程等属性。

Executor 是Java中异步执行任务的框架，它提供了一种将任务提交给执行者（线程池）并进行执行的方式，以方便地控制多线程任务的执行。

Executor 框架的主要接口是 Executor 接口，它定义了一个用于执行任务的方法：

void execute(Runnable command);

RejectedExecutionHandler

JDK 自带的 RejectedExecutionHandler 实现有 4 种

AbortPolicy：丢弃任务，抛出 RunTimeException
CallerRunsPolicy：由提交任务的线程来执行任务
DiscardPolicy：丢弃这个任务，但是不抛异常
DiscardOldestPolicy：从队列中剔除最先进入队列的任务，然后再次提交任务

线程池创建的时候，如果不指定拒绝策略就默认是 AbortPolicy 策略。

RejectedExecutionHandler 接口也可以自定义，如将任务存在数据库或者缓存中，这样可以从数据库或者缓存中获取被拒绝掉的任务。

底层机制

执行流程

刚创建出来的线程池中只有一个构造时传入的阻塞队列，里面并没有线程，如果想要在执行之前创建好核心线程数，可以调用 prestartAllCoreThreads 方法来实现，默认是没有线程的。
当有线程通过 execute 方法提交了一个任务时，首先会去判断当前线程池的线程数是否小于核心线程数，也就是线程池构造时传入的参数 corePoolSize。
- 如果小于，那么就直接通过 ThreadFactory 创建一个线程（而不是复用已有的线程）来执行这个任务。当任务执行完之后，线程不会退出，而是会去阻塞队列中获取任务。
- 如果不小于，尝试将任务放入阻塞队列中。
尝试将任务放入阻塞队列后，判断任务是否入队成功。
- 若成功，线程池调用阻塞的线程执行任务
- 若失败，判断当前线程池里的线程数是否小于最大线程数，也就是入参时的 maximumPoolSize 参数
若当前线程池里的线程数小于最大线程数，线程池会创建非核心线程来执行提交的任务，注意是优先处理这个提交的任务，而不是从队列中获取已有的任务执行。先提交的任务不一定先执行。
若当前线程池里的线程数不小于最大线程数，此时就会执行拒绝策略，即构造线程池的时候，传入的 RejectedExecutionHandler 对象，来处理这个任务。

![[ThreadPool Working Process.png]]

线程池实现复用的原理

线程在线程池内部被封装成了一个 Worker 对象，Worker 类继承了 AQS（AbstractQueuedSynchronizer 的缩写，即抽象队列同步器，是 Java.util.concurrent 中的一个基础工具类），具有一定锁的特性。

![[ThreadPool Worker class.png]]

在创建 Worker 对象的时候，会把线程和任务一起封装到 Worker 内部，然后调用 runWorker 方法来让线程执行任务

runWorker 方法源码：
![[runWorker Source Code.png]]

runWorker() 内部使用了 while 死循环，当第一个任务执行完之后，会不断地通过 getTask 方法获取任务，只要能获取到任务，就会调用 run 方法继续执行任务，这就是线程能够复用的主要原因。

但如果从 getTask 获取不到方法的话，就会调用 finally 中的 processWorkerExit 方法，将线程退出。

在执行任务之前都会调用 Worker 的 lock 方法，执行完任务之后，会调用 unlock 释放锁，这样就可以通过调用 Woker 的 tryLock 方法，根据其加锁状态判断出当前线程是否正在执行任务。

在调用 shutdown 方法关闭线程池的时候，就时用这种方式来判断线程有没有在执行任务，如果没有的话，会尝试打断没有执行任务的线程。

线程池的状态

线程池状态存储在 ctl 成员变量中的，ctl 中不仅存储了线程池的状态还存储了当前线程池中线程数的大小

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

线程池内部有 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;

RUNNING：线程池创建时就是这个状态，能够接收新任务，以及对已添加的任务进行处理。
SHUTDOWN：调用 shutdown 方法，线程池就会转换成 SHUTDOWN 状态，此时线程池不再接收新任务，但能继续处理已添加的任务到队列中。
STOP：调用 shutdownNow 方法，线程池就会转换成 STOP 状态，不接收新任务，也不能继续处理已添加的任务到队列中任务，并且会尝试中断正在处理的任务的线程。
TIDYING：SHUTDOWN 状态下，任务数为 0，其他所有任务已终止，线程池会变为 TIDYING 状态；

线程池在 SHUTDOWN 状态，任务队列为空且执行中任务为空，线程池会变为 TIDYING 状态；

线程池在 STOP 状态，线程池中执行中任务为空时，线程池会变为 TIDYING 状态。
TERMINATED：线程池彻底终止。线程池在 TIDYING 状态执行完 terminated() 方法就会转变为 TERMINATED 状态。

![[ThreadPool States.png]]

获取任务

线程在执行完任务之后，会继续调用 getTask() 方法获取任务，获取不到任务，线程会退出。

getTask 方法源码：
![[getTask source code.png]]

超时退出机制

在 getTask() 方法中，下面的代码用来判断当前获取任务的线程是否可以超时退出。

boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

如果 allowCoreThreadTimeOut 设置为 true 或者线程池当前的线程数大于核心线程数，那么该获取任务的线程就可以超时退出。

根据是否允许超时来选择调用阻塞队列 workQueue 的 poll 方法或者 take 方法。

Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();

如果允许超时，则调用 poll 方法，传入 keepAliveTime（构造线程池时传入的空闲时间），从队列中阻塞 keepAliveTime 时间来获取任务，获取不到就会返回 null，而一旦 getTask 方法返回 null，线程就会退出。
如果不允许超时，就会调用 take 方法，这个方法会一直阻塞获取任务，直到从队列中获取到任务为止。

![[getTask Working Process.png]]

关闭线程池

关闭一个线程池是一个重要的操作，它可以确保线程池中的线程正常终止并释放相关资源。

在使用线程池的过程中，及时关闭线程池可以避免资源泄漏和意外的线程执行。因此，建议在不再需要使用线程池时及时关闭它。

shutdown

shutdown() 方法：平缓地关闭线程池，使线程池进入 SHUTDOWN 状态。它会停止接受新的任务，尝试中断所有闲置的工作线程，并尝试将已经提交但尚未执行的任务执行完毕。已经在执行的任务会继续执行。
shutdown 方法源码：
![[shutDown source code.png]]

ExecutorService executor = Executors.newFixedThreadPool(10);
// 执行一些任务...
executor.shutdown();

shutdownNow

shutdownNow() 方法：尝试立即关闭线程池，使线程池进入 STOP 状态。它会停止接受新的任务，尝试打断所有的线程，并且尝试中断正在执行的任务。已经提交但尚未执行的任务会被移出阻塞队列。

这个方法返回一个 List，其中包含那些尚未执行的任务。

shutdownNow 方法源码：

![[shutdownNow resource code.png]]

ExecutorService executor = Executors.newFixedThreadPool(10);
// 执行一些任务...
List<Runnable> unexecutedTasks = executor.shutdownNow();

awaitTermination

awaitTermination(long timeout, TimeUnit unit) 方法：等待一段时间来判断线程池是否已经完全关闭。它会阻塞调用线程，直到超过指定的等待时间或者线程池完全关闭。

ExecutorService executor = Executors.newFixedThreadPool(10);
// 执行一些任务...
executor.shutdown();
try {if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {// 如果在指定的等待时间内线程池没有完全关闭，则进行其他操作}
} catch (InterruptedException e) {// 处理中断异常Thread.currentThread().interrupt();
}

监控线程池

在使用线程池的过程中，监控线程池状态是非常重要的，它可以帮助我们更好地了解线程池的运行情况和性能瓶颈，并及时发现、定位和解决问题。

监控本身也会带来一定的性能开销，因此需要在监控和性能之间进行取舍，不要过度监控从而影响性能。同时，及时对发现的问题进行解决，以保证线程池的正常运行。

Java Management Extensions

JMX：Java Management Extensions（简称 JMX）是一个为管理应用程序提供的标准框架。线程池提供了一些 MBean（即可管理的 Java 对象），可以通过 JMX Thead Pool MBean 来监测、管理线程池。可以使用 JConsole 或者其他 JMX 客户端来连接线程池，查看固定间隔时间内线程池的执行情况等等。