15-Executor框架

电玩女神 2021-08-29 18:23 348阅读 0赞

### Executor框架 ###

*  1. 简介
 *   *  1.1 Executor框架的两级调度模型
     *  1.2 Executor框架的结构与成员
     *   *  1.2.1 Executor 框架的结构
         *  1.2.2 Executor 框架的成员
 *  2. ThreadPoolExecutor 详解
 *   *  2.1 SingleThreadExecutor 详解
     *  2.2 FixedThreadPool 详解
     *  2.3 CachedThreadPool 详解
 *  3. ScheduleThreadPoolExecutor 详解
 *   *  3.1 运行机制
     *  3.2 实现原理
 *  4. FutureTask 详解
 *   *  4.1 简介
     *  4.2 实现原理

# 1. 简介 #

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

在`HotSpot JVM`的线程模型中，`Java`线程被一对一映射为本地操作系统线程。

`Java`线程启动时会创建一个本地操作系统线程；当该`Java`线程终止时，这个操作系统线程也会被回收。

在上层，`Java`多线程程序通常把应用分解为若干任务，然后由用户级的调度器（`Executor`）将这些任务映射为固定数量的线程；

在底层，操作系统内核将这些线程映射到硬件处理器上。

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

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

### 1.2.1 Executor 框架的结构 ###

`Executor`框架主要由3大部分组成：

1.  任务。包括被执行任务需要实现的接口：`Runnable`接口或`Callable`接口
2.  任务的执行。包括任务执行机制的核心接口`Executor`，以及继承自`Executor`的`ExecutorService`接口。
3.  异步计算的结果。包括接口`Future` 和实现`Future`接口的`FutureTask`类。

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

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

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

### 1.2.2 Executor 框架的成员 ###

（1）**ThreadPoolExecutor**

`ThreadPoolExecutor`通常使用工厂类`Executors`来创建。

`Executors`可以创建`3`种类型的`ThreadPoolExecutor`：

**`SingleThreadExecutor`、`FixedThreadPool`和`CachedThreadPool`**。

1. `SingleThreadExecutor`：创建使用单个线程的`SingleThreadExecutor` 的`API`：

public static ExecutorService newSingleThreadExecutor()
        
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)

适用于需要保证顺序地执行各个任务；并且在任意时间点，不会有多个线程是活动的应用场景。

2.`FixedThreadPool` ：创建使用固定线程数的 `FixedThreadPool`的`API`。

public static ExecutorService newFixedThreadPool(int nThreads)
        
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)

适用于为了满足资源管理的需求，需要限制当前线程数量的应用场景，适用于负载比较重的服务器。

3.`CachedThreadPool`：创建一个会根据需要创建新线程的`CachedThreadPool`的`API`。

public static ExecutorService newCachedThreadPool()
        
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory)

是大小无界的线程池，适用于执行很多的短期异步任务的程序或负载较轻的服务器。

（2）**ScheduledThreadPoolExecutor**

1.`ScheduleThreadPoolExecutor`：包含若干个线程的`ScheduledThreadPoolExecutor`

2.`SingleThreadScheduleExecutor`：只包含一个线程的`ScheduledThreadPoolExecutor`

`ScheduledThreadPoolExecutor`适用于需要多个后台线程执行周期任务，同时为了满足资源管理需求而需要限制后台线程的数量的应用场景。

`SingleThreadScheduleExecutor` 适用于需要单个后台线程执行周期任务，同时需要保证顺序地执行各个任务地应用场景。

（3）**Future 接口**

`Future`接口和实现`Future`接口的`FutureTask`类用来表示异步计算地结果。

（4）**Runnable接口和Callable接口**

可以使用工厂类`Executors`把一个`Runnable`包装成一个`Callable`。

public static Callable<Object> callable(Runnable task)
    
    public static <T> Callable<T> callable(Runnable task, T result)

# 2. ThreadPoolExecutor 详解 #

它是线程池的实现类。

public class ThreadPoolExecutor extends AbstractExecutorService { 
        
        private volatile int corePoolSize;
        private volatile int maximumPoolSize;
        private final BlockingQueue<Runnable> workQueue;
        private volatile RejectedExecutionHandler handler;
        
        // ......
    }

## 2.1 SingleThreadExecutor 详解 ##

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 3]  
       可以看到，`SingleThreadExecutor` 正是由于参数`corePoolSize`及`maximumPoolSize`的值为1，所以该线程池才始终只有一个线程。

需要注意的是`LinkedBlockingQueue`并不是真的没有长度限制，其最大长度为`2^31-1`

## 2.2 FixedThreadPool 详解 ##

return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());

`keepAliveTime`设置为`0L`，意味着多余的空闲线程会被立即终止。

`FixedThreadPool` 使用无界队列作为工作队列，有如下影响：

1）当线程池中的线程数达到 `corePoolSize` 后，新任务将在无界队列中等待，因此线程池中的线程数不会超过`corePoolSize`。

2）使用无界队列，`maxnumPoolSize` 无效。

3）由于 1）和 2），`keepAliveSize` 无效。

4）不会拒绝任务。

## 2.3 CachedThreadPool 详解 ##

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 4]  
       `SynchronousQueue`是一个没有容量的阻塞队列。每个插入操作必须等待另一个线程的对应移除操作，反之亦然。

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

# 3. ScheduleThreadPoolExecutor 详解 #

主要用来在给定的延迟之后运行任务，或者定期执行任务。

## 3.1 运行机制 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 6]  
       `ScheduleThreadPoolExecutor` 为了实现周期性的执行任务，对`ThreadPoolExecutor` 做了如下的修改：

*  使用DelayQueue作为任务队列
 *  获取任务的方式不同
 *  执行周期任务后，增加了的处理

## 3.2 实现原理 ##

`ScheduledThreadPoolExecutor` 把待调度的任务（`ScheduledFutureTask`）放到一个DelayQueue中.

private class ScheduledFutureTask<V>
                extends FutureTask<V> implements RunnableScheduledFuture<V> { 
    
           
            // 表示这个任务被添加到 ScheduledThreadPoolExecutor 中的序号
            private final long sequenceNumber;
    
            // 表示这个任务将要被执行的具体时间
            private long time;
    
    		// 表示任务执行的间隔周期
            private final long period;
            
            // ......
    }

`DelayQueue`封装了一个 `PriorityQueue`，这个`PriorityQueue` 会对队列中的`ScheduledFutureTask`进行排序。排序是，`time` 小的排在前面（时间早的任务将被先执行）。如果两个`ScheduledFutureTask` 的`time`相同，就比较 `sequenceNumber`，`sequenceNumber` 小的排在前面（任务执行时间相同，先提交的任务先执行）。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 7]  
       1）线程1从 `DelayQueue`中获取已到期的 `ScheduledFutureTask`(`DelayQueue.take()` )。到期任务指 `ScheduledFutureTask`的`time`大于等于当前时间。

2）线程1 执行这个`ScheduleFutureTask`。

3）线程1修改`ScheduledFutureTask`的`time` 变量为下次将要执行的时间。

4）线程1 把这个修改`time`之后的`ScheduledFutureTask` 放回 `DelayQueue`中。

**获取任务的过程**：

public E take() throws InterruptedException { 
        // 获取锁
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try { 
            for (;;) { 
                E first = q.peek();
                if (first == null)
                    //说明队列为空调用condition.await()方法，会使得当前线程释放lock然后加入到等待队列中
                    available.await();
                else { 
                    // 如果第一个数据不为空 
                    // 获取消息体的延迟时间（getDelay() 会在消息体内重写 自定义添加延迟时间）
                    long delay = first.getDelay(NANOSECONDS);
                    
                    if (delay <= 0)
                        //如果延迟时间 小于等于0 说明已经达到了延迟时间 调用poll方法返回消息体
                        return q.poll();
                    
                    //如果延迟时间不为空的话 说明还需要等待一段时间 此时重新循环 所以将frist置为空
                    first = null; // don't retain ref while waiting
                    
                    if (leader != null)
                        available.await();
                    else { 
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try { 
                            available.awaitNanos(delay);
                        } finally { 
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally { 
            if (leader == null && q.peek() != null)
                // 唤醒睡眠的线程
                available.signal();
            lock.unlock();
        }
    }

1）获取`Lock`

2）获取周期任务

1、如果 `PriorityQueue` 为空，当前线程到 `Condition` 中等待；

2、如果 `PriorityQueue` 的头元素的`time` 时间比当前时间早，到`Condition`中等待到time时间。否则返回任务。

3、获取 `PriorityQueue` 的头元素，如果PriorityQueue 不为空，则唤醒在 Condition 中等待的所有线程

3）释放锁

**添加任务到队列**：

public boolean offer(E e) { 
        final ReentrantLock lock = this.lock;
        // 获取锁
        lock.lock();
        try { 
            // 向 PriorityQueue 添加任务
            q.offer(e);
            if (q.peek() == e) { 
                leader = null;
                // 如果添加的元素是头元素，则唤醒在 Condition 中等待的所有线程
                available.signal();
            }
            return true;
        } finally { 
            // 释放锁
            lock.unlock();
        }
    }

# 4. FutureTask 详解 #

`Future` 接口和实现 `Future`接口的`FutureTask`类，代表异步计算的结果。

## 4.1 简介 ##

![在这里插入图片描述][20200604220353379.png]

除了实现`Future`接口外，还实现了`Runnable`接口。因此`FutureTask` 可以交给`Executore`执行，也可以调用线程直接执行（`FutureTask.run()`）。根据 `FutureTask.run()`方法被执行的时机，`FutureTask` 可以由如下3种状态：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 8]  
       当`FutureTask`处于未启动状态时，执行`FutureTask.cancel()`方法，将导致调用线程阻塞。当`FutureTask`处于已完成状态时，执行`FutureTask.get()`方法将导致调用线程立即返回结果或抛出异常。

当`FutureTask`处于未启动状态时，执行 `FutureTask.cancel()`方法将导致此任务用于不会被执行。当`FutureTask`处于已启动状态时，执行 `FutureTask.cancel(true)`方法将以中断执行此任务线程的方式试图停止任务；当`FutureTask` 处于已启动状态时，执行`FutureTask.cancel(false)`方法将不会对正在运行的线程产生影响（让正在执行的任务运行完成）；当`FutureTask`处于已完成状态时，执行`FutureTask.cancel(..)`方法将返回`false`。

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

## 4.2 实现原理 ##

`FutureTask` 的实现基于 `AbstractQueuedSynchronizer（AQS）`。`AQS` 是一个同步框架，它提供通用机制来原子性管理同步状态、阻塞和唤醒线程，以及被阻塞线程的队列。

每一个基于 `AQS`实现的同步器都会包含两种类型的操作：

1、至少一个`acquire`操作。这个操作阻塞调用线程，除非直到 `AQS`的状态运行这个线程继续执行。`FutureTask` 的 `acquire`操作为`get()/get(long timeout,TimeUnit unit)`方法调用。

2、至少一个 `release`操作。这个操作改变`AQS`的状态，改变后的状态可运行一个或多个阻塞线程被解除阻塞。`FutureTask` 的 `release`操作包括`run()`方法和`cancel()`方法。

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

`Sync` 是 `FutureTask` 的内部私有类，它继承自`AQS`。创建`FutureTask`时会创建内部私有的成员对象`Sync`，`FutureTask` 所有的公有方法直接委托给了内部私有的`Sync`。

`FutureTask.get()` 方法会调用 `AQS.acquireSharedInterruptibly(int arg)`方法，执行过程如下：

1）调用 `AQS.acquireSharedInterruptibly(int arg)`方法，这个方法首先会回调在子类`Sync`中实现的`tryAcquireShared()`方法来判断 `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.compareAndState(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()`

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

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

假设开始时`FutureTask`处于未启动状态或已启动状态，等待队列中已经有`3`个线程（A、B和C）在等待。此时，线程D执行`get()`方法将导致线程D也到等待队列中取等待。

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70]: /images/20210813/b1dfcb104c7c40e192f10af7f8112e3f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 1]: /images/20210813/10d9b742ceda4ceda684e91b857dd9a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 2]: /images/20210813/e6574ce082384090861e5b89ff143fcc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 3]: /images/20210813/3634d432800746dd86940125dad1fa72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 4]: /images/20210813/063acf87cb5a4f668ade1ea0ebe5422a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 5]: /images/20210813/f928dcf5bbf04cc4a4470f3f6f5dc6d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 6]: /images/20210813/45695c587b36426f8004c52e27a5bf77.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 7]: /images/20210813/c8653f3d53b34e20871838b1eeb2963c.png
[20200604220353379.png]: /images/20210813/f346354be20641ffb7ebcd9635e7d20d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 8]: /images/20210813/bcf199dc66c346edb0ece0c3dc7232f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 9]: /images/20210813/72eab6463e5a4a60af3d34458573aeae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 10]: /images/20210813/1f749d67bddf43039db404035cd1b9d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 11]: /images/20210813/56efb0b0ed1d45b4950dc0f92ac58360.png