Java的线程既是工作单元,也是执行机制。从JDK 5开始,把工作单元与执行机制分离开
来。工作单元包括Runnable和Callable,而执行机制由Executor框架提供。

Executor 框架简介

Executor的两级调度模型

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

在上层,Java多线程程序通常把应用分解为若干个任务,然后使用用户级的调度器(Executor框架)将这些任务映射为固定数量的线程;在底层,操作系统内核将这些线程映射到硬件处理器上。这种两级调度模型的示意图如下图所示。从图中可以看出,应用程序通过Executor框架控制上层的调度;而下层的调度由操作系统内核控制,下层的调度不受应用程序的控制。

Executor框架的结构与成员

首先是结构: Executor 框架主要由三大部分组成

  • 任务。包括被执行任务需要实现的接口:Runnable接口或Callable接口。
  • 任务的执行。包括任务执行机制的核心接口Executor,以及继承自Executor的ExecutorService接口。Executor框架有两个关键类实现了ExecutorService接口(ThreadPoolExecutor和ScheduledThreadPoolExecutor)。
  • 异步计算的结果。包括接口Future和实现Future接口的FutureTask类。

下面是类和接口的简介

  • Executor是一个接口,它是Executor框架的基础,它将任务的提交与任务的执行分离开来。
1
2
3
4
public interface Executor
{
    void execute(Runnable command);
}

虽然Executor是一个简单的接口,那是它为灵活且强大的异步任务执行框架提供了基础,它提供了一种标准的方法把任务的提交过程和执行过程解耦开来.

  • ThreadPoolExecutor是线程池的核心实现类,实现了ExecutorService 接口，用来执行被提交的任务。
  • ScheduledThreadPoolExecutor是一个实现类,也实现了ExecutorService 接口，可以在给定的延迟后运行命令,或者定期执行命令。ScheduledThreadPoolExecutor比Timer更灵活,功能更强大。
  • Future接口和实现Future接口的FutureTask类,代表异步计算的结果。
  • Runnable接口和Callable接口的实现类,都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行。

使用示意图如下:

使用流程

主线程首先要创建实现Runnable或者Callable接口的任务对象。工具类Executors可以把一个Runnable对象封装为一个Callable对象(Executors.callable(Runnable task)或Executors.callable(Runnable task,Object resule))。
然后可以把Runnable对象直接交给ExecutorService执行(ExecutorService.execute(Runnable command));或者也可以把Runnable对象或Callable对象提交给ExecutorService执行(ExecutorService.submit(Runnable task)或ExecutorService.submit(Callable<T>task))。

如果执行ExecutorService.submit(...),ExecutorService将返回一个实现Future接口的对象(到目前为止的JDK中,返回的是FutureTask对象)。由于FutureTask实现了Runnable,程序员也可以创建FutureTask,然后直接交给ExecutorService执行。
最后,主线程可以执行FutureTask.get()方法来等待任务执行完成。主线程也可以执行FutureTask.cancel(boolean mayInterruptIfRunning)来取消此任务的执行。

框架成员

介绍Executor框架的主要成员:ThreadPoolExecutor, ScheduledThreadPoolExecutor、Future接口、Runnable接口、Callable接口和Executors。

  • ThreadPoolExecutor

ThreadPoolExecutor通常使用工厂类Executors来创建。Executors可以创建3种类型的ThreadPoolExecutor:SingleThreadExecutor、FixedThreadPool和CachedThreadPool。

其中,

FixedThreadPool是创建固定线程数的API,FixedThreadpool适用于为了满足资源管理的需求,而需要限制当前线程数量的应用场景,适用于负载较重的服务器.使用的队列是LinkedBlcokingQueue。

singleThreadExecutor是创建单个线程的SingleThreadExecutor的API，使用的队列也是LinkedBlockingQueue。

CachedThreadPool。创建一个会根据需要创建新线程的CachedThreadPool的API。CachedThreadPool 是大小无界的线程池,适用于执行很多的短期异步任务的小程序,或者是负载较轻的服务器.使用的是SynchronousQueue。

  • ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor通常使用工厂类Executors来创建。Executors可以创建2种类型的ScheduledThreadPoolExecutor,如下。
1)ScheduledThreadPoolExecutor。包含若干个线程的ScheduledThreadPoolExecutor。
2)SingleThreadScheduledExecutor。只包含一个线程的ScheduledThreadPoolExecutor。

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

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

  • Future接口

Future接口和实现Future接口的FutureTask类用来表示异步计算的结果。当我们把Runnable接口或Callable接口的实现类提交(submit)给ThreadPoolExecutor或
ScheduledThreadPoolExecutor时,ThreadPoolExecutor或ScheduledThreadPoolExecutor会向我们返回一个FutureTask对象。下面是对应的API。

1
2
3
<T> Future<T> submit(Callable<T> task)
<T> Future<T> submit(Runnable task,T result)
Future<> submit(Runnable task)

Java仅仅保证返回的是一个实现了Future接口的对象.返回的可能不一定是FutureTask。

  • Runnable 接口和Callable 接口

Runnable接口和Callable接口的实现类,都可以被ThreadPoolExecutor或Scheduled-
ThreadPoolExecutor执行。它们之间的区别是Runnable不会返回结果,而Callable可以返回结果。
除了可以自己创建实现Callable接口的对象外,还可以使用工厂类Executors来把一个
Runnable包装成一个Callable。

###ThreadPoolExecutor 详解

Executor框架最核心的类是ThreadPoolExecutor，它是线程池的实现类，主要由下列四个组件构成。corepool 核心线程池的大小maximusPool 最大线程池的大小Blockingqueue 用来暂存任务的工作队列rejectExecutionHandler 饱和策略

通过Executors工具类，里面有三种类型的静态方法来创建三种线程池。

FixedThreadPool

可重用固定线程数的线程池，corepoolsize 和maximuspoolsize 都被设置为创建时指定的nThread。当线程数大于corepoolsize时，keepAlivetime 为多余的空闲线程等待新任务的最长时间，超过这个时间后多余的线程将会被终止。fixedThreadPool设置的是立即终止。

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

FixedThreadPool使用无界队列LinkedBlockingQueue作为线程池的工作队列（队列的容量为Integer.MAX_VALUE）。

SingleThreadExecutor

SingleThreadExecutor 是使用单个Worker线程的Executor.SingleThreadExecutor 的corePoolSize 和maximumPoolSize 被设置为1，其他参数与FixedTheadPool一样，SingleThreadExecutor 也使用无界队列LinkedBlockingQueue作为线程池的工作队列（队列容量为Integer.MAX_VALUE）

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

CachedThreadPool 详解

CachedThreadPool是一个会根据需要创建线程的线程池，CachedThreadPool的corePoolSize被设置为0，即corePool为空，maximumPoolSize 被设置为Integer.MAX_VALUE,即maximumPool 是无界的，这里把keepAliveTime 设置为60L,意味着CachedThreadPool 中的空闲线程等待新任务的最长时间为60s, 空闲线程超过60s后将会被终止。CachedThreadPool 使用没有容量的SynchronousQueue作为线程池的工作队列，但是CachedThreadPool 的maximumPool 是无界的。这意味着，如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时，CachedThreadPool 会不断创建新线程，极端情况下，CachedThreadPool 会因为创建过多线程而耗尽CPU和内存资源。

1）首先执行SynchronousQueue.offer(Runnable task)。如果当前maximumPool中有空闲线程正在执行SynchronousQueue.poll（keepAliveTimeTimeUnit.NANOSECONDS)，那么主线程执行offer操作与空闲线程执行的poll操作配对成功，主线程把任务交给空闲线程执行，execute()方法执行完成；否则执行下面的步骤2。
2）当初始maximumPool为空，或者maximumPool中当前没有空闲线程时，将没有线程执行SynchronousQueue.poll(keepAliveTime，TimeUnit.NANOSECONDS)。这种情况下，步骤1）将失败。此时CachedThreadPool会创建一个新线程执行任务，execute()方法执行完成。
3）在步骤2中新创建的线程将任务执行完后，会执行SynchronousQueue.poll（keepAliveTime，TimeUnit.NANOSECONDS)。这个poll操作会让空闲线程最多在SynchronousQueue中等待60秒钟。如果60秒钟内主线程提交了一个新任务（主线程执行步骤1）），那么这个空闲线程将执行主线程提交的新任务；否则，这个空闲线程将终止。由于空闲60秒的空闲线程会被终止，因此长时间保持空闲的CachedThreadPool不会使用任何资源。

SynchronousQueue 是一个没有容量的阻塞队列，每个插入操作必须等待另一个线程的对应移除操作，反之亦然。CachedThreadPool 使用SynchronousQueue,把主线城提交的任务传递传递给空闲线程执行，CachedThreadPool 中任务传递的示意图如下

ScheduledThreadPoolExecutor 详解

ScheduledThreadExecutor 继承自ThreadPoolExecutor,主要用来在给定的延迟之后运行任务，或者定期执行任务，ScheduledThreadPoolExecutor 可以在构造函数中指定多个对应的后台线程数。

FutureTask 详解

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

简介

FutureTask除了实现Future接口外,还实现了Runnable接口。因此,FutureTask可以交给Executor执行,也可以由调用线程直接执行(FutureTask.run())。根FutureTask.run()方法被执行的时机,FutureTask可以处于下面3种状态。

  • 1)未启动。FutureTask.run()方法还没有被执行之前,FutureTask处于未启动状态。当创建一个FutureTask,且没有执行FutureTask.run()方法之前,这个FutureTask处于未启动状态。
  • 2)已启动。FutureTask.run()方法被执行的过程中,FutureTask处于已启动状态。
  • 3)已完成。FutureTask.run()方法执行完后正常结束,或被取消(FutureTask.cancel(…)),或执行FutureTask.run()方法时抛出异常而异常结束,FutureTask处于已完成状态。

当FutureTask 处于未启动或者已启动状态时，执行futureTask.get()方法将导致调用线程阻塞；当FutureTask 处于已完成状态时，执行FutureTask .get()方法会导致调用线程立即返回结果或抛出异常。

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

FutureTask 的使用 与实现

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

FutureTask 的实现是基于AQS的，

参考文献:Java并发编程的艺术