Java高并发编程（十二）：Executor框架

你的名字 2022-04-01 07:40 251阅读 0赞

Java中的线程既是工作单元，也是执行单元。工作单元包括Runnable和Callable，而执行单元是由Executor框架支持。

# 1. Executor框架简介 #

ExecutorsService的生命周期有三种状态：运行、关闭和已终止。Executor执行的任务有四个生命周期：创建、提交、开始和完成。

Executor 接口：

void execute(Runnable command);

下面我们来详细看一下ExecutorsService接口

public interface ExecutorService extends Executor { 
        void shutdown();
        List<Runnable> shutdownNow();
        boolean isTerminated();
        boolean isShutdown();
        boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException; 
        .....

> 1.  shutdown():将执行缓慢的关闭过程，不再接受新的任务，同时等待已经提交的任务执行完成————包括那些还未开始执行的任务.
> 2.  shutdownNow():将执行除暴的关闭过程：它将尝试取消所有运行的任务；并且不再启动队列中尚未开始执行的任务。
> 3.  isTerminated():判断ExecutorService是否进入终止状态
> 4.  awaitTermination():当前线程进入阻塞状态，并且等待ExecutorService达到终止状态
> 5.  isShutdown():判断ExecutorService是否进入关闭状态

## 1.1 Executor框架的两级调度模型 ##

在HotSpot VM的线程模型中，Java线程（java.lang.Thread）被一对一映射为本地操作系统线程。Java线程启动时会创建一个本地操作系统线程；当该Java线程终止时，这个操作系统线程也会被回收。操作系统会调度所有线程并将他们分配给可用的CPU。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70]

两级调度模型指的是：

*  在上层，用户将应用分解为若干个任务，然后用 Executor框架 将这些任务映射为固定数量的线程。
 *  在底层，操作系统内核将这些线程映射到处理器上。

## 1.2 Executor框架的结构和成员 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 1]

如上图所示，Executor框架由 3大部分组成：

*  **任务**。即执行任务需要实现的接口：Runnable 接口 或 Callable接口。
 *  **任务的执行**。任务执行机制核心接口Executor、和继承自Executor的ExecutorService接口（2个关键类实现：ThreadPoolExecutor 和 ScheduledThreadPoolExecutor）。
 *  **异步计算的结果**。即 Future接口 和 它的实现类 FutureTask类。

下面是这些类或者接口的简介：

>  *  Executor是一个接口，它是Executor框架的基础，它将任务的提交与任务的执行分离开来。
>  *  ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务。
>  *  ScheduledThreadPoolExecutor是一个实现类，可以在给定的延迟后运行命令，或者定期执行命令。ScheduledThreadPoolExecutor比Timer更灵活，功能更强大。
>  *  Future接口和实现Future接口的FutureTask类，代表异步计算的结果。
>  *  Runnable接口和Callable接口的实现类，都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行。

下面是Executor框架使用示意图：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 2]

使用流程：

1.  主线程主要创建 Runnable 和 Callable 的任务对象。调用`Executors.callable(Runnable)`可以把Runnable对象包装成Future对象。
2.  再创建一个ExecutorService的一个子类对象。
3.  调用`ExecutorService.submit(Runnable / Callable<T>)` 或 `ExecutorService.execute(Runnable)`，把任务提交给ExecutorService。
4.  调用`submit()`会返回一个FutureTask对象，主线程可调用`FutureTask.get()`使主线程等待任务执行完成，也可以调用`FutureTask.cancel(boolean)`取消任务执行。

### 1.3 Executor框架成员 ###

#### 1）Executor接口 ####

Executor接口仅仅只有一个`execute(Runnable)`方法，ExecutorService接口是继承该接口的一个子接口。

#### 2）ThreadPoolExecutor ####

线程池的核心实现类，用来执行被提交的任务。

#### 3）ScheduledThreadPoolExecutor ####

是ThreadPoolExecutor的一个子类，只是ThreadPoolExecutor的一个简单封装。可以在给定的延迟后运行命令，或者定期指定命令，比 工具类Timer更加灵活、功能更强大。

#### 4）Runnable 和 Callable接口 ####

都是任务，可以提交给ExecutorService执行。

Callable 可以返回结果 且 可抛出异常，Runnable 则不可以。调用`Executors.callable(Runnable)`可以把Runnable对象包装成Future对象。

#### 5）Future接口 ####

Future接口 和 FutureTask实现类 是用来表示异步计算的结果的。JDK 1.8为止，调用`submit()`会返回一个FutureTask对象。

#### 6）Executors工具类 ####

Executors工厂类可以，

*  **创建3三种ThreadPoolExecutor**：SinglePoolExecutor、FixedThreadPool和CachedThreadPool。
 *  还可以**创建ScheduledThreadPoolExecutor**、**SingleThreadScheduledExecutor**。

### 3.3 各类线程池 和 FutureTask详解 ###

#### 1）3种ThreadPoolExecutor类型的线程池 ####

##### 1.1) FixedThreadPool #####

目的是为了**创建固定线程数量的线程池**，**适用于负载比较重的服务器**，使用了无界队列LinkedBlockingQueue。

使用Executors工厂类创建的源代码为：

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

可知：

*  corePoolSize 和 maximumPoolSize 都被设置为 指定参数 nThreads。
 *  keepAliveTime 为0，说明多余的空闲线程会被立即终止。
 *  使用了无界队列LinkedBlockingQueue，会带来如下影响：
    
     *  由于无界队列，maximumPoolSize 和 keepAliveTime 都是无效参数；
     *  由于无界队列，线程池不会拒绝任务（即不会调用饱和策略），且线程池的线程池数量不会超过corePoolSize 。

其执行流程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 3]

1.  如果 运行中的线程数量 < corePoolSize，则创建新线程执行任务。
2.  线程池完成预热之后，将任务将入 LinkedBlockingQueue。
3.  线程执行完 1 的任务后，会循环反复从 LinkedBlockingQueue 获取任务来执行。

### 3.3 各类线程池 和 FutureTask详解 ###

#### 1）3种ThreadPoolExecutor类型的线程池 ####

##### 1.1) FixedThreadPool #####

目的是为了**创建固定线程数量的线程池**，**适用于负载比较重的服务器**，使用了无界队列LinkedBlockingQueue。

使用Executors工厂类创建的源代码为：

可知：

其执行流程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 4]

##### 1.2) SingleThreadExecutor #####

使用了无界队列LinkedBlockingQueue，**目的是创建使用单个线程的Executor**，适用于**需要保证各个任务顺序执行且不会有多个线程活动**的应用场景。

使用Executors工厂类创建的源代码为：

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

使用无界队列LinkedBlockingQueue带来的影响同上。

其执行流程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 5]

流程说明略。

##### 1.3) CachedThreadPool #####

CachedThreadPool 是一个会**根据需求创建新线程的线程池**，是一个**线程数量无界**的线程池。适用于 执行很多的短任务的小程序 或者 负载较轻的服务器。使用的是无容量的无界队列SynchronousQueue。

使用Executors工厂类创建的源代码为：

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

说明：

*  corePoolSize被设置为0，maximumPoolSize为最大整数值。说明线程池的允许线程数量是无限的。
 *  keepAliveTime被设置为60s，说明空闲线程超过60s就会被停止。
 *  SynchronousQueue是没有容量的工作队列，且maximumPool是无界的，说明 主线程提交任务速度 高于 线程处理任务的速度（来一个任务就使用 **创建的新线程 或 空闲线程** 来运行，不将任务保存在任务队列中）；由于线程数量不限制，**极端情况下，可能会导致太多线程耗尽CPU资源**。

其执行流程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 6]

流程说明：

1.  先执行`SynchronousQueue.offer(Runnable)`。如果 maximumPool中有空闲线程 执行了 `poll()`，那么与主线程执行的`offer()`配对成功，主线程则把任务直接提交给空闲线程执行，`execute()`执行完成。否则，执行步骤2.
2.  当 初始maximumPool为空 或者 maximumPool 没有空闲线程时，这时没有线程执行 `poll()`，这种情况下CachedThreadPool 会创建一个新线程执行任务，`execute()`执行完成。
3.  在 步骤2 中新创建的线程执行完任务后，会执行`poll()`，会让空闲线程最多在 SynchronousQueue 中等待60s，
    
     *  如果60s内主线程提交了任务，空闲线程会执行提交的任务。
     *  否则，空闲线程被终止。

#### 2）ScheduledThreadPoolExecutor详解 ####

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，主要用来给 **给定延迟之后运行任务，或者定期执行任务**。**适用于 需要多个后台执行周期任务 且 又需要限制线程数量** 的应用场景。

使用了无界队列 DelayQueue。

**功能与Timer类似，但区别有**：

*  ScheduledThreadPoolExecutor功能更强大。
 *  Timer类不管启动多少定时器，只会启动单个线程，而ScheduledThreadPoolExecutor可以指定多条线程。
 *  Timer执行周期任务依赖于系统时间，而ScheduledThreadPoolExecutor基于时间的延迟，因此不会因系统时间的改变而改变。

使用Executors工厂类创建的源代码为：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { 
            return new ScheduledThreadPoolExecutor(corePoolSize);
        }

说明：其中corePoolSize是指定的后台线程的基本数量（但是它的maximumPool是无界的）。

ScheduledThreadPoolExecutor的构造函数如下：

public ScheduledThreadPoolExecutor(int corePoolSize) { 
            super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                  new DelayedWorkQueue());
        }

newSingleThreadScheduledExecutor的构造函数如下：

public static ScheduledExecutorService newSingleThreadScheduledExecutor() { 
            return new DelegatedScheduledExecutorService
                (new ScheduledThreadPoolExecutor(1));
        }

##### 2.1）任务的传递 #####

ScheduledThreadPoolExecutor 的任务是直接增加到 DelayQueue队列中，然后由coolPool的线程以一定方式获取任务执行，是一次**任务传递**，而不是直接交付给线程执行，其流程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 7]

为了实现周期性的执行任务，**对ThreadPoolExecutor做了3个方面的修改**：

1.  使用DelayQueue最为任务队列。
2.  获取任务的方式不同。
3.  执行周期任务后，增加了额外的处理。

##### 2.2）任务执行步骤 #####

主线程调用 `scheduleAtFixedRate()`或 `scheduleWithFixedDelay()`提交任务(一个ScheduledFutureTask对象)直接到DelayQueue队列中。

**ScheduledFutureTask主要由3个成员变量**：

<table> 
 <thead> 
  <tr> 
   <th align="center">成员变量</th> 
   <th align="center">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">long time</td> 
   <td align="center">任务被执行的时间</td> 
  </tr> 
  <tr> 
   <td align="center">long sequenceNumber</td> 
   <td align="center">任务添加到线程池的序号</td> 
  </tr> 
  <tr> 
   <td align="center">long period</td> 
   <td align="center">任务执行的间隔周期</td> 
  </tr> 
 </tbody> 
</table>

**DelayQueue 封装了一个PriorityQueue**，这个优先队列会对任务列表的任务排序，排序规则如下：

*  time小的排在前面（即时间早的任务优先被执行）。
 *  若time相同，sequenceNumber小的排在前面（time相同，先提交的任务先执行）。

**线程执行某个任务的步骤**：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 8]

1.  `DelayQueue.take()`，线程1 从DelayQueue获取已到期的任务。到期任务指的是 time大于等于当前时间的任务。
2.  线程1执行这个任务。
3.  线程1修改任务的time变量为下次执行的时间。（根据period）
4.  线程1把这个任务放回DelayQueue中。`DelayQueue.add()`。

#### 3）FutureTask详解 ####

FutureTask 实现了2个接口：Future 、 Runnable。因此，

*  FutureTask 可以交给Executor框架执行。
 *  也可以调用`FutureTask.run()`直接启动线程运行。

调用`FutureTask.run()`有3种状态：**未启动、已启动 和 已完成**，其状态迁移图如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 9]

**在3种状态下，调用`get()`和`cancel()`方法**的执行示意图如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 10]

#### 4）FutureTask实现 ####

FutureTask的实现基于AbstractQueuedSynchronizer（以下简称为AQS）。java.util.concurrent中的很多可阻塞类（比如ReentrantLock）都是基于AQS来实现的。AQS是一个同步框架，它提供通用机制来原子性管理同步状态、阻塞和唤醒线程，以及维护被阻塞线程的队列。

AQS被作为“模板方法模式”的基础类提供给FutureTask的内部子类Sync，这个内部子类只  
需要实现状态检查和状态更新的方法即可，这些方法将控制FutureTask的获取和释放操作。具体来说，Sync实现了AQS的tryAcquireShared（int）方法和tryReleaseShared（int）方法，Sync通过这两个方法来检查和更新同步状态。

FutureTask的设计示意图如图所示。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 11]

FutureTask.get()方法会调用AQS.acquireSharedInterruptibly（int arg）方法，这个方法的执行过程如下。

> 1.  调用AQS.acquireSharedInterruptibly（int arg）方法，这个方法首先会回调在子类Sync中实现的tryAcquireShared()方法来判断acquire操作是否可以成功。acquire操作可以成功的条件为：state为执行完成状态RAN或已取消状态CANCELLED，且runner不为null。
> 2.  如果成功则get()方法立即返回。如果失败则到线程等待队列中去等待其他线程执行  
>     release操作。
> 3.  当其他线程执行release操作（比如FutureTask.run()或FutureTask.cancel（…））唤醒当前线程后，当前线程再次执行tryAcquireShared()将返回正值1，当前线程将离开线程等待队列并唤醒它的后继线程（这里会产生级联唤醒的效果，后面会介绍）。
> 4.  最后返回计算的结果或抛出异常。

FutureTask.run()的执行过程如下。

> 1.  执行在构造函数中指定的任务（Callable.call()）。
> 2.  以原子方式来更新同步状态（调用AQS.compareAndSetState（int expect，int update），设置state为执行完成状态RAN）。如果这个原子操作成功，就设置代表计算结果的变量result的值为Callable.call()的返回值，然后调用AQS.releaseShared（int arg）。
> 3.  AQS.releaseShared（int arg）首先会回调在子类Sync中实现tryReleaseShared（arg）来执行release操作（设置运行任务的线程runner为null，然会返回true）；AQS.releaseShared（int arg），然后唤醒线程同步等待队列中的第一个线程。
> 4.  调用FutureTask.done()。

当执行FutureTask.get()方法时，如果FutureTask不是处于执行完成状态RAN或已取消状态CANCELLED，当前执行线程将到AQS的线程等待队列中等待（见下图的线程A、B、C和D）。当某个线程执行FutureTask.run()方法或FutureTask.cancel（…）方法时，会唤醒线程等待队列的第一个线程（见图所示的线程E唤醒线程A）。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 12]  
假设开始时FutureTask处于未启动状态或已启动状态，等待队列中已经有3个线程（A、B和  
C）在等待。此时，线程D执行get()方法将导致线程D也到等待队列中去等待。

当线程E（运行任务的线程Runner）执行run()方法时，会唤醒队列中的第一个线程A。线程A被唤醒后，首先把自己从队列中删除，然后唤醒它的后继线程B，最后线程A从get()方法返回。线程B、C和D重复A线程的处理流程。最终，在队列中等待的所有线程都被级联唤醒并从get()方法返回。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70]: /images/20220401/d95e552b67974c738a1b32ac7ecb237c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 1]: /images/20220401/654de0fa42ed4c6485c8d6cafb45d58d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 2]: /images/20220401/6ca31eba9af14722a309d6b37c4b2388.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 3]: /images/20220401/85f1e7cfb87e4edbabf033e67cd2a29e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 4]: /images/20220401/8566e05fb5ac4475af2d99ae24330caf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 5]: /images/20220401/7fa42b467b08488bb75f295b3e18ab6d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 6]: /images/20220401/0b78fbc3c3e24d70aba258a987819f8b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 7]: /images/20220401/2d94dd3c3fed4a9a9356fd7e19cb1da9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 8]: /images/20220401/3cd7046eda2544488091d029499d49c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 9]: /images/20220401/dd1a9a2ff7714ea48942893136550315.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 10]: /images/20220401/477c7ec973a44acf8ef46a4a401efcf7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 11]: /images/20220401/465e31c9bde445028e2fea5a915153f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 12]: /images/20220401/9c035fba48f841d78ec9220b9f835c6d.png