深入理解线程池ThreadPoolExecutor

今天药忘吃喽~ 2024-03-23 15:41 74阅读 0赞

> **线程池概述**

一种线程的使用模式，使用线程池方式创建线程的变化：从 **创建线程使用，再关闭线程** 转换为 **从线程池中获取一个空闲的线程使用，使用完后将线程归还给线程池**。

> **为什么使用线程池的方式创建线程？**

系统中频繁地创建线程，当**线程过多会带来调度开销，进而影响缓存局部性和整体性能**。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这**避免了在处理短时间任务时创建与销毁线程的代价**。线程池不仅能够保证内核的充分利用，还能防止过分调度。

**线程池的优势**：线程池做的工作主要是**控制运行的线程数量**，处理过程中**将任务放入队列**，然后在线程创建后启动这些任务，**如果线程数量超过了最大数量，超出数量的线程排队等候**，等其他线程执行完毕，再从队列中取出任务来执行。

线程池优点总结为以下三点：

*  **降低资源消耗**（**线程复用**）: 通过**重复利用已创建的线程**降低线程创建和销毁造成的销耗。
 *  **提高响应速度**（**控制最大并发数**）: 当任务到达时，任务可以**不需要等待线程创建就能立即执行**。
 *  **提高线程的可管理性**（**管理线程**）: 线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

--------------------

### 1. 线程池的使用 ###

项目开发中常使用线程池的方式创建和使用线程，接下来了解一下**线程池如何使用**。

> **线程池的工作流程**

![format_png][]

**执行步骤**

1.  创建了线程池后，**线程池中的线程数为零**（可以通过`prestartAllCoreThreads`方法使得一开始就创建好核心线程数）。
2.  当调用 execute()方法添加一个请求任务时，线程池会做出如下判断：
    
     *  如果正在运行的线程数量**小于**`corePoolSize`，那么马上创建线程运行这个任务。
     *  如果正在运行的线程数量**大于或等于**`corePoolSize`，那么将这个任务放入队列。
     *  如果这个时候队列满了且**正在运行的线程数量小于**`maximumPoolSize`，那么还是要创建非核心线程立刻运行这个任务。
     *  如果队列满了且**正在运行的线程数量大于或等于**`maximumPoolSize`，那么线程池会**启动饱和拒绝策略**来处理。
3.  当一个线程完成任务时，它会从队列中取下一个任务来执行。
4.  当一个线程无事可做超过一定的时间（keepAliveTime）时，线程会判断如果当前运行的线程数大于`corePoolSize`，那么这个线程就被销毁。

> **线程池构造器的七大参数**

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
    复制代码

1.  **`corePoolSize`**：**核心线程大小**
    
     *  概述：当提交一个任务到线程池时，线程池会创建一个线程来执行任务，**即使有其他空闲线程可以处理任务也会创新线程，等到工作的线程数大于核心线程数时就不会在创建了**。如果调用了线程池的`prestartAllCoreThreads`方法，线程池会提前把核心线程都创造好，并启动。
2.  **`maximumPoolSize`**：**线程池允许创建的最大线程数**
    
     *  概述：**此值必须大于等于1**。当继续提交任务时，如果队列满了，并且已创建的线程数小于最大线程数`maximumPoolSize`，则线程池会再创建新的线程执行任务。如果使用了无界队列，那么所有的任务会加入队列，`maximumPoolSize`这个参数就不起作用了。
3.  **`keepAliveTime`**：**多余的空闲线程的存活时间**
    
     *  概述：当线程池中线程数量超过`corePoolSize`时，并且多余的线程（指除了`corePoolSize`之外的线程）空闲时间达到`keepAliveTime`时，多余线程会被销毁直到只剩下`corePoolSize`个线程为止，**如果任务很多，并且每个任务的执行时间比较短，避免线程重复创建和回收，可以调大这个时间，提高线程的利用率**。
4.  **`unit`**：**keepAliveTIme的时间单位**
    
     *  概述：可以选择的单位有天、小时、分钟、毫秒、微秒、千分之一毫秒和纳秒。类型使用枚举类`java.util.concurrent.TimeUnit`选择相应时间单位。
5.  **`workQueue`**：**任务队列（阻塞队列）**
    
     *  概述：用来保存**被提交但尚未被执行**的任务，作为缓存待处理任务的阻塞队列。下面列举不同种类的阻塞队列。
     *  `ArrayBlockingQueue`：基于数组结构的有界阻塞队列，此队列按照先进先出（FIFO）原则对元素进行排序
     *  `LinkedBlockingQueue`：基于链表结构的有界（大小默认值`Integer.MAX_VALUE`）阻塞队列，此队列按照先进先出排序元素，吞吐量通常要高于ArrayBlockingQueue。
     *  `PriorityBlockingQueue`：支持优先级排序的无界阻塞队列。
     *  `SynchronousQueue`：不存储元素的阻塞队列，也即单个元素的队列,每个插入操作必须等到另外一个线程调用移除操作，否则插入操作一直处理阻塞状态，吞吐量通常要高于LinkedBlockingQueue.
     *  `DelayQueue`：使用优先级队列实现的延迟无界阻塞队列。
     *  `LinkedTransferQueue`：基于链表结构组成的无界阻塞队列。
     *  `LinkedBlockingDeque`：基于链表结构组成的双向阻塞队列。
6.  **`threadFactory`**：**表示生成线程池中工作线程的线程工厂**
    
     *  概述：**用于创建线程，一般默认工厂`DefaultThreadFactory`即可**。可以通过自定义的线程工厂给每个新建的线程设置自定义线程名。或者使用`new CustomizableThreadFactory(name)`作为线程工厂即可设置自定义线程名。
7.  **`handler`**：**拒绝策略**
    
     *  概述：**线程池的饱和策略**。表示当阻塞队列满了，并且工作线程大于等于线程池的最大线程数`maximumPoolSize`时**如何拒绝继续提交的任务的策略**。除了JDK提供的策略，可以通过实际的场景实现`RejectedExecutionHandler`接口自定义饱和策略。下面列举JDK提供的拒绝策略。
     *  `CallerRunsPolicy`: **该拒绝策略会将任务回退到调用者，让调用者的线程执行被拒绝的任务**。适用于一般并发比较小，性能要求不高，不允许失败的场景。如果任务提交速度过快，可能导致程序阻塞，性能效率上必然的损失较大（“调用者运行”一种调节机制，**该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者，从而降低新任务的流量**）
     *  `AbortPolicy`（默认）: **丢弃任务并抛出拒绝执行`RejectedExecutionException`异常信息**。该拒绝策略作为线程池默认的拒绝策略。必须处理好抛出的异常，否则会打断当前的执行流程，影响后续的任务执行。
     *  `DiscardPolicy`: **该策略直接丢弃无法处理的任务，不予任何处理也不抛出异常**。如果允许任务丢失，这是最好的一种策略。
     *  `DiscardOldestPolicy`: **该拒绝策略会丢弃阻塞队列`workQueue`中最老（最前）的一个任务，并将新任务加入**（抛弃队列中等待最久的任务，然后把当前任务加入队列中尝试再次提交当前任务）

--------------------

#### 1.1 四种固定创建方式 ####

> 了解完线程池工作过程及参数后，介绍**四种固定的创建方式以及优缺点**。

1.  `Executors.newFixedThreadPool(int nThreads)`创建固定线程数的线程池

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    复制代码

概述：创建一个**可重用固定线程数**的线程池，以**无界队列**`LinkedBlockingQueue`方式来运行这些线程。因为使用无界队列导致最大线程数`maximumPoolSize`和多余线程空闲存活时间`keepAliveTime`以及拒绝策略`handler`等参数失效。

适用场景：可以预测对线程数量明确的业务或对线程数有严格控制的场景使用。

缺点：随着线程任务不能被执行而导致无限堆积，可能产生OOM异常（主要是允许无界队列的长度为`Integer.MAX_VALUE`，造成大量任务堆积，导致内存溢出异常）

1.  `Executors.newSingleThreadExecutor()`创建单个worker线程的线程池

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    复制代码

概述：**创建一个单线程的线程池，单一的工作线程可以保证提交任务的顺序执行**。内部同样使用无界队列作为阻塞队列，使得最大线程数`maximumPoolSize`和多余线程空闲存活时间`keepAliveTime`以及拒绝策略`handler`等参数失效。

适用场景：需要保证顺序执行各个任务，并且不会同时出现多个线程的场景。

缺点：同样因为使用无界队列可能会无限堆积，导致OOM异常（主要是允许无界队列的长度为`Integer.MAX_VALUE`，造成大量任务堆积，导致内存溢出异常）。

1.  `Executors.newCachedThreadPool()`创建可缓存的线程池

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    复制代码

概述：创建一个**可缓存线程池**，**如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程**。内部使用`SynchronousQueue`作为阻塞队列。相当于有提交任务则创建线程执行，当线程空闲超过60秒则销毁线程。

适用场景：执行短期异步任务，服务器负载压力轻，执行时间短，提交任务多的场景。

缺点：当线程任务比较耗时，又大量创建时，会导致OOM异常（主要是允许创建的线程数量为`Integer.MAX_VALUE`，造成创建大量的任务，导致内存溢出异常）。

1.  `Executors.newScheduledThreadPool(int corePoolSize)`创建定长线程池，支持定时和周期性任务执行

// 内部调用ScheduledThreadPoolExecutor
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }
    复制代码

概述：该线程池可以延迟定时执行任务，同样使用无界队列作为工作队列，提供`scheduleAtFixedRate`、`scheduleWithFixedDelay`等调用方法，可以灵活选择延迟执行任务的方式。

缺点：允许的请求队列长度为`Integer.MAX_VALUE`，可能会堆积大量的请求，从而导致 OOM。

--------------------

#### 1.2 线程池状态 ####

> 从线程池实现类的源码可以看到**线程池实现类使用`AtomicInteger`类型的ctl属性记录线程池状态和线程池数量**。ctl采用分割数据区域的方式记录线程池状态和线程池数量，前3位记录线程池状态，后29位存储线程池数量，初始化时默认是RUNNING状态，线程数为0个。

![format_png 1][]

![format_png 2][]

**线程池状态**

*  `RUNNING`：线程池会接收新任务，并处理阻塞队列中的任务
 *  `SHUTDOWN`：线程池不会接收新任务，但会处理阻塞队列中的任务
 *  `STOP`：线程不会接收新任务，也不会处理阻塞队列中的任务，而且会中断正在运行的任务
 *  `TIDYING`：所有的任务都已终止了，`workerCount`为0，线程池进入该状态后会调用`terminated()`方法进入TERMINATED 状态
 *  `TERMINATED`：`terminated()`方法调用结束后的状态

**状态转换流程图**

![format_png 3][]

--------------------

#### 1.3 线程池的任务提交方式 ####

1.  提交方式一：`execute()`

public void execute(Runnable command)
    复制代码

传入实现Runnable接口的对象参数作为任务内容提交给线程池执行，该方法无返回值。

1.  提交方式二：`submit()`

// submit()在ExecutorService中的定义
    <T> Future<T> submit(Callable<T> task);
    
    <T> Future<T> submit(Runnable task, T result);
    
    Future<?> submit(Runnable task);
    复制代码

可以传入Callable接口或者是Runnable接口的实现类作为任务内容交给线程池执行，该方法有返回值，返回值为Future类型，可以通过`get()`方法取出执行结果。

--------------------

#### 1.4 线程池关闭方式 ####

1.  关闭方式一：`shutdown()`

将线程池状态设置为SHUTDOWN状态，并且线程池不会再接收新任务，内部会将队列中提交的所有任务执行完毕后，工作线程自动退出。

1.  关闭方式二：`shutdownNow()`

将线程池状态设置为STOP状态，并且线程池会将队列中等待的任务全部移除，将正在执行的任务执行完毕后，工作线程自动退出。

> 关闭线程池底层源码操作为：遍历线程池的所有线程，然后依次调用线程的`interrput()`方法来终止线程

![format_png 4][]

--------------------

### 2. 生产中使用的自定义线程池 ###

> **为什么生产中需要自定义线程池**，而不是使用前面介绍的四种创建方式呢？

阿里巴巴Java开发手册中有明确说明：

![format_png 5][]

由于固定的创建方式可能会导致OOM异常，所以**实际生产一般自己通过 `ThreadPoolExecutor` 的 7 个参数来自定义业务需求的线程池。**

--------------------

> 实际生产中**根据不同类型的任务合理的设置线程池的参数**

根据任务的特性可分为：

*  按任务的性质可分为：**CPU密集型任务、IO密集型任务和混合型任务**
 *  任务的优先级：高、中、低
 *  任务的执行时间：长、中、短
 *  任务的依赖性：是否依赖其他的系统资源，如数据库连接。

性质不同任务可以用不同规模的线程池分开处理:

*  **CPU密集型任务**应该尽可能小的线程，如配置`cpu数量+1`个线程的线程池。
 *  **IO密集型任务**并不是一直在执行任务，不能让cpu闲着，则应配置尽可能多的线程，如配置`cup数量*2`个线程的线程池。
 *  **混合型的任务**如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这2个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。
 *  可以通过`Runtime.getRuntime().availableProcessors()`方法获取cpu数量。
 *  **优先级不同任务**可以对线程池采用**优先级队列**来处理，让优先级高的先执行。
 *  使用队列的时候**建议使用有界队列**，有界队列增加了系统的稳定性，**如果采用无界队列，任务太多的时候可能导致系统OOM，直接让系统宕机。**
 *  线程池**总线程大小**对系统的性能有一定的影响，我们的目标是希望系统能够发挥最好的性能，过多或者过小的线程数量无法有效使用机器的性能。Java Concurrency inPractice书中给出了估算线程池大小的公式：`Nthreads = Ncpu（CPU的数量） × Ucpu（目标CPU的使用率） × (1 + W/C（等待时间与计算时间的比例）)`

--------------------

> SpringBoot项目中**配置自定义线程池案例**

1.  考虑到常用的参数需要放在配置类中易于修改，故创建一个配置属性注入类

/**
     * 配置属性注入类
     * @author 兴趣使然的L
     **/
    @ConfigurationProperties(prefix = "thread")   // 以thread作为前缀进行标识
    @Data        // lombok注解
    public class ThreadPoolConfigProperties {
        // 核心线程数
        private Integer coreSize;
        // 最大线程数
        private Integer maxSize;
        // 空闲存活时间
        private Integer keepAliveTime;
        // 工作队列长度
        private Integer queueSize;
    }
    复制代码

#配置线程池
    thread.coreSize=7
    thread.maxSize=20
    thread.keepAliveTime=10
    thread.queueSize=50
    复制代码

1.  创建线程池配置类并将线程池注入Spring容器

/**
     * 线程池配置类
     * @author 兴趣使然的L
     **/
    // 开启配置绑定功能并将ThreadPoolConfigProperties注入容器
    @EnableConfigurationProperties(ThreadPoolConfigProperties.class)  
    @Configuration
    public class ThreadConfig {
    
        @Bean
        public ThreadPoolExecutor threadPoolExecutor(ThreadPoolConfigProperties pool) {
            return new ThreadPoolExecutor(
                    pool.getCoreSize(),
                    pool.getMaxSize(),
                    pool.getKeepAliveTime(),
                    TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(pool.getQueueSize()),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.AbortPolicy()
            );
        }
        
    }
    复制代码

1.  测试

需求：提交100个线程任务（并发），当前配置信息为：核心线程数为7个，最大线程数为20个，队列长度为50，按照执行流程会有30个线程任务会被拒绝策略拒绝（因为7 + 50 + 13最多只可以接收70个线程，所以会有30个线程被拒绝抛出异常）

/**
     * 测试类
     * @author 兴趣使然的L
     */
    @RestController
    public class ThreadTest {
    
        @Autowired
        private ThreadPoolExecutor executor;
    
        @GetMapping("/test")
        public void test() throws RejectedExecutionException{
            executor.execute(() -> {
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("执行完成");
            });
        }
    }
    复制代码

使用JMeter进行100个线程并发访问查看是否得到想要的结果

![format_png 6][]

可以看到异常率为30%,即有30个线程会被拒绝策略拒绝，符合预期结果。说明自定义线程池配置成功。

--------------------

> 关于获取线程池监控信息的方式

ThreadPoolExecutor 提供了一些方法用于查询线程池的当前状态、线程池大小、活动线程数量、任务数量等信息。

*  `getActiveCount()`：获取线程池中正在执行任务的线程数量
 *  `getCompletedTaskCount()`：获取线程池已完成的任务数量，该值小于等于taskCount
 *  `getCorePoolSize()`：获取线程池的核心线程数量
 *  `getLargestPoolSize()`：获取线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否满过，也就是达到了maximumPoolSize
 *  `getMaximumPoolSize()`：获取线程池的最大线程数量
 *  `getPoolSize()`：获取线程池当前的线程数量
 *  `getTaskCount()`：获取线程池已经执行的和未执行的任务总数

生产中可以配置自定义的监控线程在特定时间间隔打印线程池的信息，可以写入日志中，便于监控线程池的状态信息。

--------------------

### 3. 线程池源码理解 ###

> **线程池的`execute()`方法的核心执行过程**

1.  `execute()`方法源码

public void execute(Runnable command) {
            // 判断入参是否为空
    	if (command == null)
    		throw new NullPointerException();
            // 获取原子属性ctl（包含线程数量与线程池状态）
    	int c = ctl.get();
            // workerCountOf()用于获取线程池的当前线程数量
    	if (workerCountOf(c) < corePoolSize) {
                    // 小于核心线程数则直接调用addWorker创建新线程执行任务并返回
    		if (addWorker(command, true))
    			return;
                    // 重新获取ctl
    		c = ctl.get();
    	}
            // 走到这一步说明当前线程数量已经超过核心线程数，需要判断能否加入阻塞队列
            // 判断线程池状态是否为RUNNING并将任务放入阻塞队列
    	if (isRunning(c) && workQueue.offer(command)) {
                    // 获取ctl
    		int recheck = ctl.get();
                    // 如果线程池没有RUNNING，成功从阻塞队列中删除任务，执行reject方法处理任务
    		if (! isRunning(recheck) && remove(command))
    			reject(command);
                    // 线程池处于RUNNING状态，但是没有线程，则创建线程
    		else if (workerCountOf(recheck) == 0)
    			addWorker(null, false);
    	}
            // 如果创建新线程失败，则执行拒绝策略
    	else if (!addWorker(command, false))
    		reject(command);
    }
    复制代码

**分析：**

*  `ctl.get()`可以获取到**线程池的状态以及线程数量**。
 *  `workerCountOf(ctl)`可以将ctl的前3位和后29位拆分，**返回后29位的值（即当前线程数量）**。
 *  第一步：通过**获取到的当前线程数和核心线程数进行对比**，如果**小于则可以直接创建新线程执行任务**。
 *  如果第一步无法满足则进行第二步：通过获取到的ctl使用`isRunning(ctl)`判断线程池**是否是RUNNING状态，是则加入阻塞队列**。
    
     *  **加入成功**则对ctl进行双重检查，如果线程池没有处于RUNNING状态并且从队列中移除任务，则执行拒绝策略操作，否则如果当前线程数量为0则创建新线程执行任务。
     *  **加入失败**则说明队列已满，此时则直接创建除了核心线程以外的线程执行任务，如果创建失败，则说明可能最大线程数也满了，则执行拒绝策略操作。

1.  `addWorker()`方法源码

`execute()`方法中最主要的创建线程方法`addWorker()`。

private boolean addWorker(Runnable firstTask, boolean core) {
        // 通过CAS更新线程数量
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
    
            // Check if queue empty only if necessary.
            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;
                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
            }
        }
    
        // 创建启动线程
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    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();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
    复制代码

**分析：**

*  `addWorker()`方法用于创建新的线程并执行任务，分为两个部分**更新线程数量和创建启动线程**。
 *  **使用CAS更新线程数量**，使用`compareAndIncrementWorkerCount()`即CAS操作增加线程数量，循环更新直到成功为止。
 *  **创建启动线程部分使用全局锁（独占锁）`mainLock`创建并执行线程**，并记录状态，如果创建失败则调用`addWorkerFailed()`移除线程操作。

1.  `runWorker()`方法源码

`addWorker()`方法中的创建启动线程的`new Worker()`创建内部类`Worker`里的使用`run()`调用核心方法`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 {
            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 {
                    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);
        }
    }
    复制代码

**分析：**

*  线程启动后，通过`unlock()`方法释放锁，并将AQS的`completedAbruptly`设置为true，表示运行可中断。
 *  获取并执行任务：
    
     *  **加锁操作**，保证thread不被其他线程中断。
     *  **检查线程池状态**，如果线程池处于STOP状态则中断线程。
     *  **执行`beforeExecute()`方法**：任务执行之前调用的方法，有2个参数，第1个参数是执行任务的线程，第2个参数是任务。
     *  **执行任务的`run()`方法**
     *  **执行`afterExecute()`方法**：任务执行完成之后调用的方法，2个参数，第1个参数表示任务，第2个参数表示任务执行时的异常信息，如果无异常，第二个参数为null。
     *  **解锁操作**。

**补充：**

可以通过重写`beforeExecute()`和`afterExecute()`方法来拓展线程池

// 拓展线程池例子
    new ThreadPoolExecutor(
            pool.getCoreSize(),
            pool.getMaxSize(),
            pool.getKeepAliveTime(),
            TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(pool.getQueueSize()),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
    ){
    	@Override
    	protected void beforeExecute(Thread t, Runnable r) {
    		System.out.println("开始执行");
    		super.beforeExecute(t, r);
    	}
    
    	@Override
    	protected void afterExecute(Runnable r, Throwable t) {
    		System.out.println("结束执行");
    		super.afterExecute(r, t);
    	}
    
    	@Override
    	protected void terminated() {
    		System.out.println("关闭线程池");
    		super.terminated();
    	}
    };
    复制代码

1.  `getTask()`方法源码

`runWorker()`方法里的任务通过`getTask()`循环获取任务。

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
    
        for (;;) {
            int c = ctl.get();
    
            // Check if queue empty only if necessary.
            if (runStateAtLeast(c, SHUTDOWN)
                && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
    
            int wc = workerCountOf(c);
    
            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
    复制代码

**分析：**

*  `getTask()`方法从阻塞队列中获取任务。
 *  核心在于`timed`而`timed`在于`allowCoreThreadTimeOut || wc > corePoolSize`。
 *  **`timed`为false时，执行`workQueue.take`**,如果阻塞队列为空当前线程会被挂起，直到队列有任务加入，线程被唤醒，并从队列中获取任务执行。（没有任务会阻塞）
 *  **`timed`为true时，执行`workQueue.poll`**,如果在`keepAliveTime`存活时间内，阻塞队列还是没有任务，则返回null。（没有任务直接返回null）

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/9cde0155350b4b2f89355f4256db8b54.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/6ce93d4b635f4068ad05312a3336b0bf.png
[format_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/1854bbda7a114228af49c9743fe2743f.png
[format_png 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/4a2ff35692fd440ea8ebd421526c8498.png
[format_png 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/2f5ae3446da14ba58c12886d322a4d74.png
[format_png 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/eb664bd3179f414290086f1b2520d6fd.png
[format_png 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/3c697ff0ddec4a74be4a98bd99076175.png