Executor框架简介

Posted 张伯雨

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Executor框架简介相关的知识，希望对你有一定的参考价值。

Executor框架是在Java5中引入的，可以通过该框架来控制线程的启动，执行，关闭，简化并发编程。Executor框架把任务提交和执行解耦，要执行任务的人只需要把任务描述清楚提交即可，任务的执行提交人不需要去关心。

通过Executor框架来启动线程比使用Thread更好，更易管理，效率高，避免this逃逸问题。

Executor的实现还提供了对生命周期的支持，以及统计信息收集，应用程序管理机制和性能监视等机制。

Executor框架

源码版本：

jdk1.7.0_71

Executor框架由3大部分组成：

  1. 任务：被执行任务需要实现接口Runnable、Callable。
  2. 任务执行：任务执行机制的核心接口Executor，继承Executor的ExecutorService。CompletionService等。
  3. 异步计算的结果：Future和实现了Future接口的FutureTask类。

Executor

是一个接口，只定义了一个方法void execute(Runnable command);该方法接受一个Runnable实例，作用就是执行提交的任务，实现在子类中。

ExecutorService

也是一个接口，继承自Executor接口，提供了更多的方法，提供了生命周期的管理方法，以及可跟踪一个或多个异步任务执行状况的方法。

ExecutorService的生命周期包括三种状态：运行，关闭，终止。创建后便进入运行状态，当调用了shutdown()方法时，进入关闭状态，此时不再接受新任务，但是它还在执行已经提交了的任务，当所有的任务执行完后，便达到了终止状态。

shutdown()

关闭当前服务，当调用此方法时，它将不再接受新的任务，已经提交的任务，还要继续执行完毕。

shutdownNow()

1
List<Runnable> shutdownNow();

关闭当前服务，尚未执行的任务，不再执行；正在执行的任务，通过线程中断thread.interrupt。方法返回等待执行的任务列表。

isShutdown()

程序是否已经关闭。

isTerminated()

程序是否已经终止。已经关闭并且所有的任务都执行完成，返回true，其他返回false。

awaitTermination()

1
2
boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行。如果程序终止，返回true，如果超时，返回false，等待时发生中断，抛出异常。

submit(Callable task)

1
<T> Future<T> submit(Callable<T> task);

向Executor提交一个Clallable任务，并返回一个Future；

submit(Runnable task, T result)

1
<T> Future<T> submit(Runnable task, T result);

向Executor提交一个Runnable任务，并返回一个Future；

submit(Runnable task)

1
Future<?> submit(Runnable task);

向Executor提交一个Runnable任务，并返回一个Future；

invokeAll（）

执行所有任务列表，当所有任务执行完成之后，返回Future列表。

invokeAny()

在执行的任务集合中，任何一个完成就返回。

Executors

提供了一系列静态工厂方法用于创建各种线程池。返回的线程池都实现了ExecutorService接口。

如果没有特殊要求，请尽量使用此类中提供的静态方法生成线程池。

大部分方法的底层实现都在ThreadPoolExecutor类中，有关ThreadPoolExecutor介绍请往下翻。ThreadPoolExecutor的构造函数如下：

1
2
3
4
5
6
7
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

newFixedThreadPool(int nThreads)

创建一个可重用的固定线程数的线程池，以共享无界队列方式来运行这些线程。

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

corePoolSize和maximumPoolSize大小一样，使用无界queue的话maximumPoolSize参数是没有意义的。改方法使用的是LinkedBlockingQueue无界队列。keepAliveTime为0.

newFixedThreadPool(nThreads, ThreadFactory)

使用给定的ThreadFactory创建一个可重用的固定线程数的线程池，以共享无界队列方式运行这些线程。

newSingleThreadExecutor

创建一个使用单个线程的线程池，使用无界队列来运行该线程。

newSingleThreadExecutor(ThreadFactory)

使用给定的ThreadFactory创建一个单个线程的线程池，使用无界队列来运行该线程。

newCachedThreadPool

创建一个可根据需要创建新线程的线程池，以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言，这些线程池通常可提高程序性能。

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

maximumPoolSize为最大，是一个无界线程池。workqueue使用SynchronousQueue，该queue中每个插入操作必须等待另一个线程的对应移除操作。它的主要提供的功能是线程复用，但不能控制线程数量。

newCachedThreadPool(ThreadFactory)

使用给定的ThreadFactory创建一个可根据需要创建新线程的线程池，以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言，这些线程池通常可提高程序性能。

newSingleThreadScheduledExecutor

创建一个使用单个线程的线程池，该线程池支持定时以及周期性执行任务的功能。

newScheduledThreadPool(corePoolSize)

创建一个指定数量的线程的线程池，该线程池支持定时以及周期性执行任务的功能。

AbstractExecutorService

ExecutorService方法的默认实现类。

ScheduledExecutorService

一个可定时调度任务的接口。扩展了ExecutorService。支持Future和定期执行任务。

ScheduledThreadPoolExecutor

ScheduledExecutorService的实现，一个可定时调度任务的线程池。

ThreadPoolExecutor

线程池，可以通过调用Executors的静态工厂方法来创建线程池并返回一个ExecutorService对象。

ThreadPoolExecutor是Executors的底层实现，看Executors源码可知，大部分的生成线程池的方法内部都是调用此类的方法。

该类继承了AbstractExecutorService。

构造方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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使用无界队列LinkedBlockingQueue等时，此参数无效。
  • keepAliveTime当前线程池线程总数大于核心线程数时，终止多余的空闲线程时间。
  • unitkeepAliveTime参数的时间单位

  • workQueue工作队列，如果当前线程池达到核心线程数时，且当前所有线程都处于活动状态，则将新加入的任务放到此队列中。仅保持由execute方法提交的Runnable任务。

    • ArrayBlockingQueue 基于数组结构有界队列，FIFO原则对任务进行排序，队列满了之后的任务，调用拒绝策略。
    • LinkedBlockingQueue 基于链表结构的无界队列，FIFO原则对任务进行排序。
    • SynchronousQueue 直接将任务提交给线程而不是将它加入到队列，实际上此队列是空的。每个插入的操作必须等到另一个调用移除的操作；如果新任务来了线程池没有任何可用线程处理的话，则调用拒绝策略。
    • PriorityBlockingQueue 具有优先级的队列的有界队列，可以自定义优先级；默认是按自然排序。

  • threadFactory 执行程序创建新线程时使用的工厂。

  • handler 超出线程范围和队列容量时，执行被阻塞，此时可以选择用此handler进行处理。

Callable，Future与FutureTask

在之前创建线程的时候都是继承Thread或者实现Runnable接口，这种方法在任务完成后无法获取执行结果。而在Java5之后有了Runnable和Future，可以在任务执行完之后得到任务执行结果。

Callable

Callable能产生结果，Future能拿到结果。Callable和Runnable接口类似，Runnable不会返回结果，也无法抛出返回结果的异常，而Callable可以返回值，这个值可被Future拿到。

Callable一般和ExecutorService一起使用，ExecutorService的submit方法提交的是Runnable或者是Callable类型的Task。

Future

Future对具体提交的Callable任务执行结果进行取消，查询是否完成，获取结果。可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。Future接口的具体实现都在FutureTask中。

cancel()方法

1
boolean cancel(boolean mayInterruptIfRunning);

可以取消任务，成功返回true，失败返回false。

参数mayInterruptIfRunning表示是否取消正在执行但是没有执行完成的任务，true可以取消，false不取消

  • 如果任务已经完成，无论参数是true或false，都返回false
  • 如果任务正在执行，若参数为true，返回true；若参数为false，返回false；
  • 如果任务还未执行，无论参数是true或false，都返回true。

isCancelled()方法

表示任务是否被取消成功。

isDone()方法

表示任务是否已经完成。

Java并发Executor框架

java多线程之Executor框架线程池详细介绍与ThreadPoolExecutor

2，Executor线程池

）

Executor框架Executors工厂类

Java线程池Executor框架详解