Java并发类库中线程池的使用及其工作原理

怼烎@ 2022-05-18 08:34 107阅读 0赞

## 一、Java 并发类库提供的线程池有哪几种？ 分别有什么特点？ ##

通常开发者都是利用 Executors 提供的通用线程池创建方法，去创建不同配置的线程池，主要区别在于不同的ExecutorService 类型或者不同的初始参数。

**Executors 目前提供了 5 种不同的线程池创建配置：**  
（1）、`newCachedThreadPool()`，它是一种用来处理大量短时间工作任务的线程池，具有几个鲜明特点：它会试图缓存线程并重用，当无缓存线程可用时，就会创建新的工作线程；如果线程闲置的时间超过 60 秒，则被终止并移出缓存；长时间闲置时，这种线程池，不会消耗什么资源。其内部使用 SynchronousQueue 作为工作队列。

（2）、`newFixedThreadPool(int nThreads)`，重用指定数目（nThreads）的线程，其背后使用的是无界的工作队列，任何时候最多有 nThreads 个工作线程是活动的。这意味着，如果任务数量超过了活动队列数目，将在工作队列中等待空闲线程出现；如果有工作线程退出，将会有新的工作线程被创建，以补足指定的数目 nThreads。

（3）、`newSingleThreadExecutor()`，它的特点在于工作线程数目被限制为 1，操作一个无界的工作队列，所以它保证了所有任务的都是被顺序执行，最多会有一个任务处于活动状态，并且不允许使用者改动线程池实例，因此可以避免其改变线程数目。

（4）、`newSingleThreadScheduledExecutor()`和`newScheduledThreadPool(int corePoolSize)`，创建的是个 `ScheduledExecutorService`，可以进行定时或周期性的工作调度，区别在于单一工作线程还是多个工作线程。

（5）、`newWorkStealingPool(int parallelism)`，这是一个经常被人忽略的线程池，Java 8 才加入这个创建方法，其内部会构建ForkJoinPool，利用Work-Stealing算法，并行地处理任务，不保证处理顺序。

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

**TIP:**另外，线程池这个定义就是个容易让人误解的术语，因为 ExecutorService 除了通常意义上“池”的功能，还提供了更全面的线程管理、任务提交等方法。

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

## 二、Executor 框架的基本组成 ##

在大多数应用场景下，使用`Executors` 提供的 5 个静态工厂方法就足够了，但是仍然可能需要直接利用 `ThreadPoolExecutor`等构造函数创建，这就要求你对线程构造方式有进一步的了解。

首先来了解`Executor`框架的基本组成：  
![这里写图片描述][70]

**分析上图中的各个组成部分：**  
1、`Executor`是一个基础的接口，其初衷是将任务提交和任务执行细节解耦，这一点可以体会其定义的唯一方法。

void execute(Runnable command);

2、`ExecutorService`则更加完善（可以理解为`Executor` 的升级版），不仅提供 `service`的管理功能，比如 `shutdown`等方法，也提供了更加全面的提交任务机制，如返回`Future`而不是 `void`的 `submit`方法。

3、Java 标准类库提供了几种线程池的实现，比如`ThreadPoolExecutor`、`ScheduledThreadPoolExecutor`、`ForkJoinPool`。这些线程池的设计特点在于其高度的可调节性和灵活性，以尽量满足复杂多变的实际应用场景。

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

## 三、线程池内部工作原理 ##

![这里写图片描述][70 1]

**（1）、**工作队列负责存储用户提交的各个任务，这个工作队列，可以是容量为 0 的 SynchronousQueue（如： `newCachedThreadPool`），也可以是 `LinkedBlockingQueue`（如`newFixedThreadPool`）；

private final BlockingQueue<Runnable> workQueue;

**（2）、**内部的“线程池”，这是指保持工作线程的集合，线程池需要在运行过程中管理线程创建、销毁。例如，对于带缓存的线程池，当任务压力较大时，线程池会创建新的工作线程。当业务压力退去，线程池会在闲置一段时间后结束线程；

//线程池的工作线程被抽象为静态内部类 Worker，基于AQS实现。
    private final HashSet<Worker> workers = new HashSet<>();

**（3）、**`ThreadFactory`提供上面所需要的创建线程逻辑；

**（4）、**如果任务提交时被拒绝，比如线程池已经处于 `SHUTDOWN`状态，需要为其提供处理逻辑，Java 标准库提供了类似`ThreadPoolExecutor、AbortPolicy`等默认实现，也可以按照实际需求自定义。

为了能对线程生命周期有个更加清晰的印象，可以参考一个简单的状态流转图，对线程池的可能状态和其内部方法之间进行了对应（注意，实际 Java 代码中并不存在所谓 Idle 状态，我添加它仅仅是便于理解。）

![这里写图片描述][70 2]

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

## 四、线程池的基本组成部分 ##

从上面的分析，就可以看出线程池的几个基本组成部分，基本都体现在线程池的构造函数中。如：

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

**(1)、**`corePoolSize`:所谓的核心线程数，可以大致理解为长期驻留的线程数目（除非设置了 allowCoreThreadTimeOut）。对于不同的线程池，这个值可能会有很大区别，比如 newFixedThreadPool 会将其设置为 nThreads，而对于 newCachedThreadPool 则是为 0。  
**(2)、**`maximumPoolSize`，顾名思义，就是线程不够时能够创建的最大线程数。同样进行对比，对于 newFixedThreadPool，当然就是 nThreads，因为其要求是固定大小，而 newCachedThreadPool 则是 Integer.MAX\_VALUE。  
**(3)、**`keepAliveTime`和 `TimeUnit`，这两个参数指定了额外的线程能够闲置多久，显然有些线程池不需要它。  
**(4)、**`workQueue`，工作队列，必须是 `BlockingQueue`。  
通过配置不同的参数，我们就可以创建出行为大相径庭的线程池，这就是线程池高度灵活性的基础。

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

## 五、典型的 `execute`方法源码分析 ##

public void execute(Runnable command) {
    …
        int c = ctl.get();
        // 检查工作线程数目，低于 corePoolSize 则添加 Worker
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
       // isRunning 就是检查线程池是否被 shutdown
       // 工作队列可能是有界的，offer 是比较友好的入队方式
        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);
        }
        // 尝试添加一个 worker，如果失败以为着已经饱和或者被 shutdown 了
        else if (!addWorker(command, false))
            reject(command);
    }

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

## 五、线程池的使用问题 ##

#### 1、注意事项： ####

**①、避免任务堆积。**前面我说过 `newFixedThreadPool`是创建指定数目的线程，但是其工作队列是无界的，如果工作线程数目太少，导致处理跟不上入队的速度，这就很有可能占用大量系统内存，甚至是出现 OOM。诊断时，你可以使用 jmap 之类的工具，查看是否有大量的任务对象入队。  
**②、避免过度扩展线程。**  
**③如果线程数目不断增长（可以使用 jstack 等工具检查），也需要警惕另外一种可能性，就是线程泄漏。**  
**④、避免死锁等同步问题。**  
**⑤、尽量避免在使用线程池时操作 `ThreadLocal`**

#### 2、线程池大小的选择策略: ####

①、如果我们的**任务主要是进行计算**，那么就意味着 CPU 的处理能力是稀缺的资源，我们能够通过大量增加线程数提高计算能力吗？往往是不能的，如果线程太多，反倒可能导致大量的上下文切换开销。所以，这种情况下，通常建议按照 **CPU 核的数目 N 或者 N+1**。

②、如果是需要较多等待的任务，例如 I/O 操作比较多，可以参考：  
`线程数 = CPU 核数 × （1 + 平均等待时间 / 平均工作时间）`

上面是仅仅考虑了 CPU 等限制，实际还可能受各种系统资源限制影响。不要把解决问题的思路全部指望到调整线程池上，很多时候架构上的改变更能解决问题。

[70]: /images/20220518/7bde2baa3a1241588b903a3d03eea779.png
[70 1]: /images/20220518/299d018861ff42a189b12d8c47a233ef.png
[70 2]: /images/20220518/811b0dbeaa574c1fb51886836e07440e.png