Java并发编程之六--线程池

约定不等于承诺〃 2023-05-30 11:58 43阅读 0赞

# 一、什么是线程池？为什么使用线程池？ #

**线程池介绍：**

线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。如果某个线程在托管代码中空闲（如正在等待某个事件），则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。

**使用线程池可以解决的问题：**

1.  创建/销毁线程伴随着系统开销，过于频繁的创建/销毁线程，会很大程度上影响处理效率
    
    例如：  
    记创建线程消耗时间T1，执行任务消耗时间T2，销毁线程消耗时间T3  
    如果T1+T3>T2，那么是不是说开启一个线程来执行这个任务太不划算了！  
    正好，线程池缓存线程，可用已有的闲置线程来执行新任务，避免了T1+T3带来的系统开销  
    降低资源的消：降低线程的创建和销毁时资源的消耗；
2.  线程并发数量过多，抢占系统资源从而导致阻塞
    
    我们知道线程能共享系统资源，如果同时执行的线程过多，就有可能导致系统资源不足而产生阻塞的情况  
    运用线程池能有效的控制线程最大并发数，避免以上的问题
3.  对线程进行一些简单的管理，提高线程的可管理性  
    比如：延时执行、定时循环执行的策略等，运用线程池都能进行很好的实现

# 二、线程池的使用 #

在 Java 中，新建一个线程池对象非常简单，Java 本身提供了工具类`java.util.concurrent.Executors`，可以使用如下代码创建一个固定数量线程的线程池：

ExecutorService service = Executors.newFixedThreadPool(10);

注意：以上代码用来测试还可以，实际使用中最好能够显示地指定相关参数。

我们可以看下其内部源码实现：

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

在阿里巴巴代码规范中，建议我们自己指定线程池的相关参数，为的是让开发人员能够自行理解线程池创建中的每个参数，根据实际情况，创建出合理的线程池。接下来，我们来剖析下`java.util.concurrent.ThreadPoolExecutor`的构造方法参数。

## ThreadPoolExecutor 浅析 ##

`java.util.concurrent.ThreadPoolExecutor`有多个构造方法，我们拿参数最多的构造方法来举例，以下是阿里巴巴代码规范中给出的创建线程池的范例：

ThreadPoolExecutor service = new ThreadPoolExecutor(5, 200,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<>(1024), 
                    new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build(), 
                    new ThreadPoolExecutor.AbortPolicy());

其构造函数如下：

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler) {
            if (corePoolSize < 0 ||
                maximumPoolSize <= 0 ||
                maximumPoolSize < corePoolSize ||
                keepAliveTime < 0)
                throw new IllegalArgumentException();
            if (workQueue == null || threadFactory == null || handler == null)
                throw new NullPointerException();
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }

首先最重要的几个参数，可能就是：`corePoolSize`，`maximumPoolSize`，`workQueue`了，先看下这几个参数的解释：

*  int corePoolSize  
    用于设定 thread pool 需要时刻保持的最小 core threads 的数量，即便这些 core threads 处于空闲状态啥事都不做也不会将它们回收掉，当然前提是你没有设置 allowCoreThreadTimeOut 为 true。至于 pool 是如何做到保持这些个 threads 不死的，我们稍后再说。
 *  int maximumPoolSize  
    用于限定 pool 中线程数的最大值。如果你自己构造了 pool 且传入了一个 Unbounded 的 queue 且没有设置它的 capacity，那么不好意思，最大线程数会永远 <= corePoolSize，maximumPoolSize 变成了无效的。
 *  long keepAliveTime  
    线程空闲下来后，存活的时间，这个参数只在> corePoolSize才有用，超过这个时间，多余的线程会被回收。
 *  TimeUnit unit    
    存活时间的单位值
 *  BlockingQueue<Runnable> workQueue  
    该线程池中的任务队列：维护着等待执行的 Runnable 对象。当所有的核心线程都在干活时，新添加的任务会被添加到这个队列中等待处理，如果队列满了，则新建非核心线程执行任务
 *  ThreadFactory threadFactory  新线程的产生方式
 *  RejectedExecutionHandler handler 拒绝策略

这些参数中，比较容易引起问题的有`corePoolSize`, `maximumPoolSize`, `workQueue`以及`handler`：

*  `corePoolSize`和`maximumPoolSize`设置不当会影响效率，甚至耗尽线程；
 *  `workQueue`设置不当容易导致OOM；
 *  `handler`设置不当会导致提交任务时抛出异常。

正确的参数设置方式会在下文给出。

## 线程池的工作顺序、工作原理 ##

If fewer than corePoolSize threads are running, the Executor always prefers adding a new thread rather than queuing.
    If corePoolSize or more threads are running, the Executor always prefers queuing a request rather than adding a new thread.
    If a request cannot be queued, a new thread is created unless this would exceed maximumPoolSize, in which case, the task will be rejected.

corePoolSize -> 任务队列 -> maximumPoolSize -> 拒绝策略

线程池的处理流程如下：

1.  判断线程池里的核心线程是否都在执行任务，如果不是（核心线程空闲或者还有核心线程没有被创建）则创建一个新的工作线程来执行任务。如果核心线程都在执行任务，则进入下个流程。
2.  线程池判断工作队列是否已满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
3.  判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5_size_16_color_FFFFFF_t_70 1][]

## **合理配置和使用线程池** ##

### **1.合理配置线程池参数** ###

根据任务的性质来：计算密集型（CPU），IO密集型，混合型

计算密集型：加密，大数分解，正则…….， 线程数适当小一点，最大推荐：机器的Cpu核心数+1，为什么+1，防止页缺失，(机器的Cpu核心=Runtime.*getRuntime*().availableProcessors();)

IO密集型：读取文件，数据库连接，网络通讯, 线程数适当大一点，机器的Cpu核心数\*2,

混合型：尽量拆分，IO密集型>>计算密集型，拆分意义不大，IO密集型~计算密集型

队列的选择上，应该使用有界，无界队列可能会导致内存溢出，OOM

### **2.合理使用线程池** ###

**避免使用无界队列**

不要使用`Executors.newXXXThreadPool()`快捷方法创建线程池，因为这种方式会使用无界的任务队列，为避免OOM，我们应该使用`ThreadPoolExecutor`的构造方法手动指定队列的最大长度：

ExecutorService executorService = new ThreadPoolExecutor(2, 2, 
                    0, TimeUnit.SECONDS, 
                    new ArrayBlockingQueue<>(512), // 使用有界队列，避免OOM
                    new ThreadPoolExecutor.DiscardPolicy());

**明确拒绝任务时的行为**

任务队列总有占满的时候，这是再`submit()`提交新的任务会怎么样呢？`RejectedExecutionHandler`接口为我们提供了控制方式，接口定义如下：

public interface RejectedExecutionHandler {
        void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
    }

线程池给我们提供了几种常见的拒绝策略：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5_size_16_color_FFFFFF_t_70 2][]

<table> 
 <thead> 
  <tr> 
   <th>拒绝策略</th> 
   <th>拒绝行为</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>AbortPolicy</td> 
   <td>抛出RejectedExecutionException</td> 
  </tr> 
  <tr> 
   <td>DiscardPolicy</td> 
   <td>什么也不做，直接忽略</td> 
  </tr> 
  <tr> 
   <td>DiscardOldestPolicy</td> 
   <td>丢弃执行队列中最老的任务，尝试为当前提交的任务腾出位置</td> 
  </tr> 
  <tr> 
   <td>CallerRunsPolicy</td> 
   <td>直接由提交任务者执行这个任务</td> 
  </tr> 
 </tbody> 
</table>

线程池默认的拒绝行为是`AbortPolicy`，也就是抛出`RejectedExecutionHandler`异常，该异常是非受检异常，很容易忘记捕获。如果不关心任务被拒绝的事件，可以将拒绝策略设置成`DiscardPolicy`，这样多余的任务会悄悄的被忽略。当然我们也可以自定义拒绝策略，只需实现RejectedExecutionHandler接口即可，而且在生产环境也经常使用这种拒绝策略，方便记录日志或做数据持久化以便后面再次尝试执行被拒绝的任务。

ExecutorService executorService = new ThreadPoolExecutor(2, 2, 
                    0, TimeUnit.SECONDS, 
                    new ArrayBlockingQueue<>(512), 
                    new ThreadPoolExecutor.DiscardPolicy());// 指定拒绝策略

## 预定义的线程池 ##

*  newSingleThreadExecutor 创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
    
        public static ExecutorService newSingleThreadExecutor() {
                return new FinalizableDelegatedExecutorService
                    (new ThreadPoolExecutor(1, 1,
                                            0L, TimeUnit.MILLISECONDS,
                                            new LinkedBlockingQueue<Runnable>()));
            }
    
     
 *  newFixedThreadPool  创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。
    
        public static ExecutorService newFixedThreadPool(int nThreads) {
                return new ThreadPoolExecutor(nThreads, nThreads,
                                              0L, TimeUnit.MILLISECONDS,
                                              new LinkedBlockingQueue<Runnable>());
            }
    
    corePoolSize和maximumPoolSize的大小是一样的（实际上，如果使用无界queue的话，maximumPoolSize参数是没有意义的）  
    keepAliveTime和unit的值：说明该实现不想keep alive  
    最后的BlockingQueue选择了LinkedBlockingQueue，该queue有一个特点，它是无界的。
 *  newCachedThreadPool 创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以添加新线程来处理任务。此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者JVM）能够创建的最大线程数。
    
        public static ExecutorService newCachedThreadPool() {
                return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                              60L, TimeUnit.SECONDS,
                                              new SynchronousQueue<Runnable>());
            }
    
    无界线程池，可以进行自动线程回收。  
    无界线程池的maximumPoolSize为bigInteger。BlockingQueue使用的是SynchronousQueue。  
    该Queue中，每个插入操作必须等待另一个线程的对应移除操作。
 *  newScheduledThreadPool  创建一个大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。
    
        public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
                return new ScheduledThreadPoolExecutor(corePoolSize);
            }

对于BlockingQueue workQueue，一共有三种类型的queue。  
所有BlockingQueue都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互。

如果运行的线程少于corePoolSize，则Executor始终首选添加新的线程，而不进行排队。  
（如果当前运行的线程小于corePoolSize，则任务根本不会存放，添加到queue中，而是直接开始运行）  
如果运行的线程等于或多于corePoolSize，则Executor始终首选将请求加入队列，而不添加新的线程。  
如果无法将请求加入队列，则创建新的线程，除非创建此线程超出maximumPoolSize，在这种情况下，任务将被拒绝.

## 排队有三种通用策略： ##

1.直接提交：

> 工作队列的默认选项是SynchronousQueue，它将任务直接提交给线程而不保持它们。    
>   在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。    
>   此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。    
>   直接提交通常要求无界maximumPoolSizes以避免拒绝新提交的任务。    
>   当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

2.无界队列：

> 使用无界队列（例如，不具有预定义容量的LinkedBlockingQueue）将导致在所有corePoolSize线程都忙时新任务在队列中等待。    
>   这样，创建的线程就不会超过corePoolSize。（因此，maximumPoolSize的值也就无效了。）    
>   当每个任务独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在web服务器中。    
>   这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

3.有界队列：

> 当使用有限的maximumPoolSizes时，有界队列（如ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。

队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低CPU使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。  
如果任务频繁阻塞（例如，它们是I/O边界），则系统可能为超过许可的更多线程安排时间。  
使用小型队列通常要求较大的池大小，CPU使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

## **BlockingQueue的选择** ##

例一：使用直接提交策略，即SynchronousQueue。

> 首先SynchronousQueue是无界的，即存储任务的能力是没有限制的，但是由于该Queue本身的特性，在某次添加元素后必须等待其他线程取走后才能继续添加。  
> 在这里不是核心线程便是新创建的线程。  
> 使用以下参数构造ThreadPoolExecutor：
> 
>     new ThreadPoolExecutor(
>         2, 3, 30, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
>         new RecorderThreadFactory("CookieRecorderPool"),
>         new ThreadPoolExecutor.CallerRunsPolicy()
>     );
> 
> 当核心线程已经有2个正在运行。  
> 1. 此时，继续来了一个任务（A），根据前面介绍的“如果运行的线程等于或多于corePoolSize，则Executor始终首选将请求加入队列，而不添加新的线程”，所以A被添加到queue中。  
> 2. 又来了一个任务（B），且核心2个线程还没有忙完，会先尝试加入队列，但是由于使用的是SynchronousQueue，所以一定无法加入进去。  
> 3. 此时便满足了上面提到的“如果无法将请求加入队列，则创建新的线程，除非创建此线程超出maximumPoolSize，在这种情况下，任务将被拒绝。”所以必然会创建一个线程来运行这个任务。  
> 4. 如果这三个任务都还没完成，连续来了两个任务，第一个加入queue中，后一个到来时，queue中无法插入，而线程数达到了maximumPoolSize，所以只好执行异常策略了。  
> 所以在使用SynchronousQueue通常要求maximumPoolSize是无界的，这样就可以避免上述情况发生（如果希望限制就直接使用有界队列）。  
> 对于使用SynchronousQueue的作用：此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。  
> 即，任务A1，A2有内部关联，A1需要先运行，那么先提交A1，再提交A2，当使用SynchronousQueue时可以保证，A1必定先被执行前，A2不可能添加入queue中。

例二：使用无界队列策略，即LinkedBlockingQueue。

> 这个就拿newFixedThreadPool来说，根据前文提到的规则：  
> 如果运行的线程少于corePoolSize，则Executor始终首选添加新的线程，而不进行排队。  
> 如果运行的线程多于或等于corePoolSize，则Executor始终首选将请求加入队列，而不添加新的线程。  
> 如果无法将请求加入队列，则创建新的线程，除非创建此线程超过maximumPoolSize，这时，任务将被拒绝。。  
> 不像SynchronousQueue那样有其自身的特点，对于无界队列来说，总是可以加入的（资源耗尽另说）。即，永远也不会触发产生新的线程！  
> corePoolSize大小的线程数会一直运行，忙完当前的，就从队列中拿任务开始运行。所以要防止任务疯长，比如任务运行的时间较长，而添加任务的速度远远超过处理任务的时间，而且还不断增加，没多久就爆了。

例三：有界队列，使用ArrayBlockingQueue。

> 这个是最为复杂 的使用，JDK不推荐使用，与上面的相比，最大的特点是可以防止资源耗尽。

本文借鉴：[Java线程池理解][Java]

源码文字解读可参考[由浅入深理解Java线程池及线程池的如何使用][Java 1]

源码视频解读可参考[Java高并发六、线程池][Java 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191110133600330.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191110133608223.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191110134303505.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWR1XzQxNjY5OTE5,size_16,color_FFFFFF,t_70
[Java]: https://blog.csdn.net/ymrfzr/article/details/51479816
[Java 1]: https://cloud.tencent.com/developer/article/1089881
[Java 2]: https://www.bilibili.com/video/av73397362/