Java 并发编程 Executor 框架

本文部分摘自《Java 并发编程的艺术》

Excutor 框架

1. 两级调度模型

在 HotSpot VM 的线程模型中，Java 线程被一对一映射为本地操作系统线程。在上层，Java 多线程程序通常应用分解成若干个任务，然后使用用户级的调度器（Executor）将这些任务映射为固定数量的线程；在底层，操作系统内核将这些线程映射到硬件处理器。这种两级调度模型的示意图如图所示：

从图中可以看出，应用程序通过 Executor 框架控制上层调度，下层的调度则由操作系统内核控制

2. 框架结构

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

• 任务

  包括被执行任务需要实现的接口：Runnable 接口或 Callable 接口

• 任务的执行

  包括任务执行机制的核心接口 Executor，以及继承自 Executor 的 ExecutorService 接口。Executor 框架有两个关键类实现了 ExecutorService 接口，分别是 ThreadPoolExecutor 和 ScheduleThreadPoolExecutor，它们都是线程池的实现类，可以执行被提交的任务

• 异步计算的结果

  包括接口 Future 和实现 Future 接口的 FutureTask 类

3. 执行过程

主线程首先要创建实现 Runnable 或 Callable 接口的任务对象，可以使用工具类 Executors 把一个 Runnable 对象封装为一个 Callable 对象

// 返回结果为 null
Executors.callable(Runnable task);
// 返回结果为 result
Executors.callable(Runnable task, T result);

然后把 Runnable 对象直接交给 ExecutorService 执行

ExecutorService.execute(Runnable command);

或者把 Runnable 对象或 Callbale 对象提交给 ExecutorService 执行

ExecutorService.submit(Runnable task);
ExecutorService.submit(Callable<T> task);

如果执行 ExecutorService.submit 方法，将会返回一个实现 Future 接口的对象 FutureTask。最后，主线程可以执行 FutureTask.get() 方法来等待任务执行完成，也可以执行 FutureTask.cancel(boolean mayInterruptIfRunning) 来取消此任务的执行

ThreadPoolExecutor

Executor 框架最核心的类是 ThreadPoolExecutor，它是线程池的实现类，有关介绍可以参考之前写过的一篇文章

下面分别介绍三种 ThreadPoolExecutor

1. FixedThreadPool

FixedThreadPool 被称为可重用固定线程数的线程池，下面是 FixedThreadPool 的源代码实现

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

FixedThreadPool 的 corePoolSize 和 maximumPoolSize 都被设置为创建 FixedThreadPool 时指定的参数 nThreads。当线程池中的线程数大于 corePoolSize 时，keepAliveTime 为多余的空闲线程等待新任务的最长时间，超过这个时间后多余的线程将被终止。这里把 keepAliveTime 设置为 0L，意味着多余的空闲线程会被立即终止

FixedThreadPool 的 execute() 运行示意图如下所示

对上图说明如下：

• 如果当前运行的线程少于 corePoolSize，则创建新线程来执行任务
• 线程池完成预热之后（当前运行的线程数等于 corePoolSize），将任务加入 LinkedBlockingQueue
• 线程执行完 1 中的任务后，会在循环中反复从 LinkedBlockingQueue 获取任务来执行

FixedThreadPool 使用*队列 LinkedBlockingQueue 作为线程池的工作队列（队列的容量为 Integer.MAX_VALUE），使用*队列作为工作队列会对线程池带来如下影响：当线程池中的线程数达到 corePoolSize 后，新任务将在*队列中等待，而*队列几乎可以容纳无限多的新任务，因此线程池中的线程数永远不会超过 corePoolSize，因此 maximumPoolSize 就成了无效参数，keepAliveTime 也是无效参数，运行中的 FixThreadPool 不会拒绝任务

2. SingleThreadExecutor

SingleThreadExecutor 是使用单个 worker 线程的 Executor，下面是 SingleThreadExecutor 的源代码实现

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

SingleThreadExecutor 的 corePoolSize 和 maximumPoolSize 被设置为 1，其他参数与 FixedThreadPool 相同。SingleThreadExecutor 使用*队列 LinkedBlockingQueue 作为线程池的工作队列，其带来的影响与 FixedThreadPool 相同，这里就不再赘述了

对上图说明如下：

• 如果当前运行的线程数少于 corePoolSize（即线程池中无运行的线程），则创建一个新线程来执行任务
• 在线程池完成预热之后（当前线程池中有一个运行的线程），将任务加入 LinkedBlockingQueue
• 线程执行完 1 中的任务后，会在一个无限循环中反复从 LinkedBlockingQueue 获取任务来执行

3. CachedThreadPool

CachedThreadPool 是一个会根据需要创建新线程的线程池，下面是创建 CachedThreadPool 的源代码

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

CachedThreadPool 的 corePoolSize 被设置为 0，即 corePool 为空。maximumPoolSize 被设置为 Integer.MAX_VALUE，即 maximumPool 是*的。这里把 keepAliveTime 设置为 60L，意味着 CachedThreadPool 中的空闲线程等待新任务的最长时间为 60 秒，空闲线程超过 60 秒后将会被终止

CachedThreadPool 使用没有容量的 SynchronousQueue 作为线程池的工作队列，但 CachedThreadPool 的 maximumPool 是*的。这意味着，如果主线程提交任务的速度高于 maximumPool 中线程处理任务的速度，CachedThreadPool 会不断创建新线程。极端情况下，CachedThreadPool 会因为创建过多线程而耗尽 CPU 和内存资源

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor 会把待调度的任务（ScheduledFutureTask）放到一个 DelayQueue 中。ScheduledFutureTask 主要包含三个成员变量

• long 型成员变量 time，表示这个任务将要被执行的具体时间
• long 型成员变量 sequenceNumber，表示这个任务被添加到 ScheduledThreadPoolExecutor 中的序号
• long 型成员变量 period，表示任务执行的间隔周期

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

下图是 ScheduledThreadPoolExecutor 中的线程执行周期任务的过程

• 线程 1 从 DelayQueue 获取已到期的 ScheduledFutureTask，到期任务是指 ScheduledFutureTask 的 time 大于等于当前时间
• 线程 1 执行这个 ScheduledFutureTask
• 线程 1 修改 ScheduledFutureTask 的 time 变量为下次将要被执行的时间
• 线程 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) {
                available.await();
            } else {
                long delay = first.getDelay(TimeUnit.NANOSECONDS);
                if (delay > 0) {
                    long tl = available.awaitNanos(delay);
                } else {
                    E x = q.poll();
                    assert x != null;
                    if (q.size() != 0)
                        available.signalAll();
                    return x;
                }
            }
        }
    } finally {
        lock.unlock();
    }
}

获取任务分为三大步骤：

• 获取 Lock
• 获取周期任务
  • 如果 PriorityQueue 为空，当前线程到等待队列中等待，否则执行下面的步骤
  • 如果 PriorityQueue 的头元素的 time 时间比当前时间大，到等待队列等待 time 时间，否则执行下面的步骤
  • 获取 PriorityQueue 的头元素，如果 PriorityQueue 不为空，则唤醒在等待队列中等待的所有线程
• 释放 Lock

ScheduledThreadPoolExecutor 在一个循环中执行步骤二，直到线程从 PriorityQueue 获取到一个元素之后才会退出无限循环

最后我们再看把任务放入 DelayQueue 的过程，下面是源码实现

public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        E first = q.peek();
        q.offer(e);
        if (first == null || e.compareTo(first) < 0) {
            available.signalAll();
        }
        return true;
    } finally {
        lock.unlock();
    }
}

添加任务分为三大步骤：

• 获取 Lock
• 添加任务
  • 向 PriorityQueue 添加任务
  • 如果添加的任务是 PriorityQueue 的头元素，唤醒在等待队列中等待的所有线程
• 释放 Lock

FutureTask

1. 简介

Future 接口和实现 Future 接口的 FutureTask 类，代表异步计算的结果。FutureTask 除了实现 Future 接口外，还实现了 Runnable 接口。因此，FutureTask 可以交给 Executor 执行，也可以由调用线程直接执行 FutureTask.run()。根据 FutureTask.run() 方法被执行的时机，FutureTask可以处于下面三种状态：

• 未启动

  FutureTask.run() 方法还没有被执行之前，FutureTask 处于未启动状态，当创建一个 FutureTask，且没有执行 FutureTask.run() 方法之前，这个 FutureTask 处于未启动状态

• 已启动

  FutureTask.run() 方法被执行的过程中，FutureTask 处于已启动状态

• 已完成

  FutureTask.run() 方法执行完后正常结束，或被取消 FutureTask.cancel(…)，或执行 FutureTask.run() 方法时抛出异常而结束，FutureTask 处于已完成状态

下图是 FutureTask 的状态迁移图

下图是 get 方法和 cancel 方法的执行示意图

• 当 FutureTask 处于未启动或已启动状态时，执行 FutureTask.get() 方法将导致调用线程阻塞
• 当 FutureTask 处于已完成状态时，执行 FutureTask.get() 方法将导致调用线程立即返回结果或抛出异常
• 当 FutureTask 处于未启动状态时，执行 FutureTask.cancel() 方法将导致此任务永远不会被执行
• 当 FutureTask 处于已启动状态时，执行 FutureTask.cancel(true) 方法将以中断执行此任务线程的方式来试图停止任务
• 当 FutureTask 处于已启动状态时，执行 FutureTask.cancel(false) 方法将不会对正在执行此任务的线程产生影响（让正在执行的任务运行完成）
• 当 FutureTask 处于已完成状态时，执行 FutureTask.cancel(…) 方法将返回 false

2. 使用

可以把 FutureTask 交给 Executor 执行，也可以通过 ExecutorService.submit(...) 方法返回一个 FutureTask，然后执行 FutureTask.get() 方法或 FutureTask.cancel(...) 方法，还可以单独使用 FutureTask

当一个线程需要等待另一个线程把某个任务执行完后它才能继续执行，此时可以使用 FutureTask。假设有多个线程执行若干任务，每个任务最多只能被执行一次。当多个线程试图同时执行同一个任务时，只允许一个线程执行任务，其他线程需要等待这个任务执行完后才能继续执行

private final ConcurrentMap<Object, Future<String>> taskCache = new ConcurrentHashMap<>();

private String executionTask(final String taskName)
        throws ExecutionException, InterruptedException {
    while (true) {
        Future<String> future = taskCache.get(taskName); // 1.1, 2.1
        if (future == null) {
            Callable<String> task = new Callable<String>() {
                @Override
                public String call() throws InterruptedException {
                    return taskName;
                }
            };
            FutureTask<String> futureTask = new FutureTask<String>(task);
            future = taskCache.putIfAbsent(taskName, futureTask); // 1.3
            if (future == null) {
                future = futureTask;
                futureTask.run(); // 1.4 执行任务
            }
        }
        try {
            return future.get(); // 1.5, 2.2
        } catch (CancellationException e) {
            taskCache.remove(taskName, future);
        }
    }
}

上述代码的执行示意图如图所示：

• 两个线程试图同时执行同一个任务，这里使用了线程安全的 ConcurrentHashMap 作为任务缓存可能到了注释
• 两个线程都执行到 // 1.1, 2.1 这行时，假设线程一首先得到 future，根据接下来的代码可得知，线程一创建任务放入缓存，并执行，而线程二获取线程一创建的任务，不需创建
• 两个线程都在 // 1.5, 2.2 处等待结果，只有线程一执行完任务后，线程二才能从 future.get() 返回

3. 实现

FutureTask 的实现基于 AbstractQueuedSynchronizer（AQS）

FutureTask 声明了一个内部私有的继承 AQS 的子类 Sync，对 FutureTask 所有公有方法的调用都会委托给这个内部子类，FutureTask 的设计示意图如下所示：

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

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

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

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

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

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

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