线程池ThreadPoolExecutor

系统管理员 2022-03-19 02:58 494阅读 0赞

> 主要流程如下，如果看懂了，那么这篇就可以跳过了。

![image][]  
图片来源：[https://www.cnblogs.com/KingJack/p/9595621.html][https_www.cnblogs.com_KingJack_p_9595621.html]

> 下面是线程池内容的详情

#### 核心内容：`ThreadPoolExecutor` ####

一般来说使用线程池的方式，可以通过工具类`Executors`，比如：

Executors.newSingleThreadExecutor();  // 单个线程的线程池
    Executors.newFixedThreadPool(5);   // 固定大小线程池
    Executors.newCachedThreadPool();   // 无边界线程池

我们来看他们的具体实现：

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    
    
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    
    
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

他们具体实现都是通过创建一个`ThreadPoolExecutor`类的对象，我们先来看这个类的完整构造方法：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
    
    
    corePoolSize : 核心线程数
    maximumPoolSize : 最大线程数
    keepAliveTime : 最大空闲等待回收时间
    unit : 时间单位（时，分，秒。。）
    workQueue : 阻塞队列
    threadFactory : 线程工厂
    handler : 拒绝策略

### 官方代码中的注释说明： ###

线程池解决了两个不同的问题：它们通常能够在执行大量异步任务时提供改进的性能，因为减少了每个任务的调用开销，它们提供了一种绑定和管理资源的方法，包括执行任务集合时消耗的线程。 每个`ThreadPoolExecutor`还维护一些基本统计信息，例如已完成任务的数量。为了在各种上下文中有用，这个类提供了许多可调参数和可扩展性钩子。但是，程序员被要求使用更方便的`Executors`工厂方法：

`Executors#newCachedThreadPool`（无界线程池，带自动线程回收）  
`Executors#newFixedThreadPool`（固定大小的线程池）  
`Executors#newSingleThreadExecutor`（单一后台线程的线程池）

上面几种方式预配置了最常见用法场景的设置。否则，在手动配置和调整此类时，请使用以下指南：

#### 核心线程数和最大线程数 ####

`ThreadPoolExecutor`将自动调整池大小， 根据`corePoolSize`和`maximumPoolSize`设置的边界。 当在方法`execute（Runnable)`中提交新任务时，并且当前池中运行的线程数少于`corePoolSize`线程，即使其他工作线程处于空闲状态，也会创建一个新线程来处理请求。如果有多个`corePoolSize`的工作线程，但运行的线程少于`maximumPoolSize`，则只有队列已满时才会创建新线程。通过设置`corePoolSize`和`maximumPoolSize`相同，可以创建固定大小的线程池（比如`newFixedThreadPool()`和`newSingleThreadExecutor()`）。通过将`maximumPoolSize`设置为基本无界限的值（例如`Integer.MAX_VALUE`），您可以允许池容纳任意数量的并发任务。最典型的情况是，核心和最大池大小仅在构造时设置，但也可以用 `#setCorePoolSize`和`#setMaximumPoolSize`动态更改它们。

#### 按需构建 ####

默认情况下，即使核心线程最初也是在新任务到达时创建和启动，但可以使用方法`prestartCoreThread()`（预先启动一个核心线程）或`prestartAllCoreThreads()`（预先启动所有核心线程）。如果您使用非空队列构造池，则可能需要预启动线程。

#### 创建新线程 ####

使用`ThreadFactory`创建新线程。如果没有另外指定，则使用 `Executors#defaultThreadFactory`，它创建的线程都在同一个`ThreadGroup`中，并具有相同的`NORM_PRIORITY`优先级和非守护进程状态。通过提供不同的ThreadFactory，您可以更改线程的名称，线程组，优先级，守护程序状态等。

#### 保持活动时间 ####

如果池当前有多个`corePoolSize`线程，多余的线程将被终止，如果它们的空闲时间超过`keepAliveTime`，这提供了一种在不积极使用池时减少资源消耗的方法。如果池稍后变得更活跃，则将构造新线程。也可以使用方法`setKeepAliveTime`动态更改此参数。使用值`Long.MAX_VALUE TimeUnit#NANOSECONDS`可以有效地禁止空闲线程在关闭之前永久终止。默认情况下，仅当存在多个`corePoolSize`线程时，保持活动策略才适用。但方法`allowCoreThreadTimeOut（boolean）`也可用于将此超时策略应用于核心线程，只要`keepAliveTime`值为非零即可。

#### 排队 ####

任何`BlockingQueue`都可用于转移和保留提交的任务。此队列的使用与池大小对应调整：

1.  如果运行的线程少于`corePoolSize`，则`Executor`总是更喜欢添加新线程，而不是排队。
2.  如果`corePoolSize`或更多线程正在运行，则`Executor`总是更喜欢排队请求而不是添加新的线程。
3.  如果请求无法排队，则创建新线程，除非这将超过`maximumPoolSize`，在这种情况下，任务将被拒绝。

排队有三种常规策略：

##### 1. 直接切换。 #####

工作队列的一个很好的默认选择是`SynchronousQueue`，将任务交给线程而不是另外持有他们。在这里，如果没有线程立即可用于运行它，则尝试对任务进行排队将失败，因此将构造一个新线程。当处理可能具有内部依赖性的请求集时，此策略可避免锁定。 直接切换通常需要无限制的`maximumPoolSizes`以避免拒绝新提交的任务。这反过来承认，当请求平均到达的速度超过可以处理的速度时，会导致无限制线程增长的可能性。

##### 2. 无界队列。 #####

使用无界队列（例如`LinkedBlockingQueue`没有预定义容量），将导致新任务在队列中等待`corePoolSize`核心线程忙完继续处理。因此，只会创建`corePoolSize`线程。 （并且`maximumPoolSize`的值因此没有任何影响，也就是没意义，但是要注意的是不能设置最大线程数小于核心线程数，否则抛参数异常。当每个任务完全独立于其他任务时，这可能是适当的，因此任务不能影响彼此的执行。

##### 3. 有界队列。 #####

有界队列（例如，`ArrayBlockingQueue`）有助于在使用有限`maximumPoolSizes`时防止资源耗尽，但可能更难调整和控制。队列大小和最大池大小可以互相对应，使用大队列和小池最小化CPU使用率，操作系统资源和上下文切换开销，但可能导致人为的低吞吐量。如果任务经常阻塞（例如，如果它们是I / O绑定的话），系统可能能够安排时间以获得比您允许的更多线程。使用小队列通常需要更大的池大小，这会使CPU更加繁忙但可能会遇到不可接受的调度开销，这也会降低吞吐量。所以如何定线程数和队列大小需要根据实际可能的场景来。

#### 拒绝策略 ####

当Executor关闭时，方法`execute（Runnable）`中提交的新任务将被拒绝，当`Executor`对最大线程和工作队列容量使用有限边界时，并且已经饱和。在任何一种情况下，`execute()`方法都会调用其`RejectedExecutionHandler`的`RejectedExecutionHandler#rejectedExecution（Runnable,ThreadPoolExecutor）`方法。提供了四个预定义的处理程序策略

*  `ThreadPoolExecutor.AbortPolicy`

在默认的`ThreadPoolExecutor.AbortPolicy`中，处理程序在rejection时抛出运行时`RejectedExecutionException`。

*  `ThreadPoolExecutor.CallerRunsPolicy`  
    在`ThreadPoolExecutor.CallerRunsPolicy`中，调用`execute()`本身的线程运行任务。这提供了一种简单的反馈控制机制，可以降低新任务提交的速度。
 *  `ThreadPoolExecutor.DiscardPolicy`  
    在`ThreadPoolExecutor.DiscardPolicy`中，简单地删除了拒绝执行的任务。
 *  `ThreadPoolExecutor.DiscardOldestPolicy`  
    在`ThreadPoolExecutor.DiscardOldestPolicy`中，如果`Executor`未关闭，则会删除工作队列头部的任务，然后重试执行（可能会再次失败），导致重复此操作。

#### 钩子方法 ####

此类提供`protected overridable beforeExecute(Thread，Runnable)`和`afterExecute(Runnable，Throwable)`方法在执行每个任务之前和之后调用。这些可以用来操纵执行环境;例如，重新初始化ThreadLocals，收集统计信息或添加日志条目。 此外，可以重写方法`terminated()`以执行`Executor`完全终止后需要执行的任何特殊处理。如果钩子或回调方法抛出异常，内部worker线程可能会失败并突然终止。

#### 源码分析 ####

来看看最终执行任务处理的源码，也就是`execute()`方法

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            /*
              *执行的流程实际上分为三步
              *1、如果运行的线程小于corePoolSize，以用户给定的Runable对象新开一个线程去执行
              *  并且执行addWorker方法会以原子性操作去检查runState和workerCount，以防止当返回false的
              *  时候添加了不应该添加的线程
              *2、 如果任务能够成功添加到队列当中，我们仍需要对添加的线程进行双重检查，有可能添加的线程在前
              *  一次检查时已经死亡，又或者在进入该方法的时候线程池关闭了。所以我们需要复查状态，并有有必
              *  要的话需要在停止时回滚入列操作，或者在没有线程的时候新开一个线程
              *3、如果任务无法入列，那我们需要尝试新增一个线程，如果新建线程失败了，我们就知道线程可能关闭了
              *  或者饱和了，就需要拒绝这个任务
              */
            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);
        }

#### 总结： ####

1.  比如固定线程池newFixedThreadPool(5)，只要提交5个请求，不管是否连续，比如提交一个请求，执行完了之后，再提交一个请求，它也不会使用之前创建好，且当前是空闲的线程，还是会创建新线程处理，直到创建到当前池中的线程数等于核心线程池数量为止。
2.  当设定核心线程数，最大线程数，有界队列时

new ThreadPoolExecutor(5, 10, 30L, TimeUnit.SECONDS,
                                    new LinkedBlockingQueue<Runnable>(5))

假设不断添加请求，首先不停的创建线程处理，也就是前5个请求，导致创建了5个核心线程，发现当前池中已经创建线程数=核心线程数，那么后续请求，丢到`new LinkedBlockingQueue<Runnable>(5)`这个队列中，这个时候如果有核心线程中有空闲会去拉取队列中请求进行处理，如果后续还有请求过来，发现队列满了，那么才会创建新的线程，直到池中线程数等于最大线程数，也就是10个，如果达到了最大线程数之后，此时队列也是满的，那么实行拒绝策略，默认`ThreadPoolExecutor.AbortPolicy`，表示后续请求直接抛出异常，不进行处理，直到可以创建新线程或者可以放入队列中。

这里我指定的保持活动时间为30s，即池中的线程过了30s都没有需要处理的请求，那么销毁线程，直到当前池中的线程数等于核心线程数，如果还有设置`allowCoreThreadTimeOut(true)`，那么会将核心线程也按照过期时间销毁，直到池中线程数为0

1.  我们看`newCachedThreadPool()`的实现

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());

核心线程数0，最大线程数为Integer.MAX\_VALUE，保持活动过期时间为60s，队列为`SynchronousQueue`，这个队列可以看做是容量为1的队列，所以当有请求提交到线程池的时候，每次都会创建新的线程，而老的线程执行完交给它的请求后，过60s自动销毁。

#### TIPS ####

##### 1. 使用`newCachedThreadPool()`能支持创建`Integer.MAX_VALUE`个线程吗，或者通过指定最大线程数`maximumPoolSize`，线程池能支持无限个线程吗？ #####

No，根据源码表示所支持运行中的线程数量为(2^29)-1，大约为5亿

In order to pack them into one int, we limit workerCount to  (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2 billion) otherwise representable. If this is ever an issue in the future, the variable can be changed to be an AtomicLong, and the shift/mask constants below adjusted. But until the need arises, this code is a bit faster and simpler using an int.
    
    private static final int COUNT_BITS = Integer.SIZE - 3;

##### 2. 线程池参数配置技巧 #####

可以参考：[https://www.cnblogs.com/waytobestcoder/p/5323130.html][https_www.cnblogs.com_waytobestcoder_p_5323130.html]

[image]: /images/20220319/5cb124baeadd47328c212ff4d65ea8a0.png
[https_www.cnblogs.com_KingJack_p_9595621.html]: https://www.cnblogs.com/KingJack/p/9595621.html
[https_www.cnblogs.com_waytobestcoder_p_5323130.html]: https://www.cnblogs.com/waytobestcoder/p/5323130.html