【Java多线程】Executor框架的详解

在Java中，使用线程来异步执行任务。Java线程的建立与销毁须要必定的开销，若是咱们为每个任务建立一个新线程来执行，这些线程的建立与销毁将消耗大量的计算资源。同时，为每个任务建立一个新线程来执行，这种策略可能会使处于高负荷状态的应用最终崩溃。

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

Executor框架简介

Executor框架的两级调度模型

在HotSpot VM的线程模型中，Java线程被一对一映射为本地操做系统线程。Java线程启动时会建立一个本地操做系统线程；当Java线程终止时，这个操做系统线程也会被回收。操做系统会调用全部线程并将他们分配给可用的CPU。

能够将此种模式分为两层，在上层，Java多线程程序一般把应用程序分解为若干任务，而后使用用户级的调度器（Executor框架）讲这些任务映射为固定数量的线程；在底层，操做系统内核将这些线程映射到硬件处理器上。

两级调度模型的示意图：

从图中能够看出，该框架用来控制应用程序的上层调度（下层调度由操做系统内核控制，不受应用程序的控制）。

 

Executor框架的结构和成员

Executor框架的结构

1. 任务

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

2. 任务的执行

包括任务执行机制的核心接口Executor，以及继承自Executor的ExecutorService接口。

Executor框架有两个关键类实现了ExecutorService接口：ThreadPoolExecutor 和 ScheduledThreadPoolExecutor

3. 异步计算的结果

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

Executor框架的类与接口

示意图

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

 

Executor框架的使用

 先来看个图：

1. 主线程首先要建立实现 Runnable接口或者Callable接口的任务对象。工具类Executors能够把一个Runnable对象封装为一个Callable对象
   

   Executors.callable(Runnale task); 
   或
   Executors.callable(Runnable task, Object resule); 

2. 而后能够把Runnable对象直接交给ExecutorService执行
   

   ExecutorServicel.execute(Runnable command);
   或者也能够把Runnable对象或Callable对象提交给ExecutorService执行
   ExecutorService.submit(Runnable task);

   若是执行ExecutorService.submit(...)，ExecutorService将返回一个实现Future接口的对象(到目前为止的JDK中，返回的是FutureTask对象)。因为FutureTask实现了Runnable接口，咱们也能够建立FutureTask类，而后直接交给ExecutorService执行。　　

3. 最后，主线程能够执行FutureTask.get()方法来等待任务执行完成。主线程也能够执行FutureTask.cancel(boolean mayInterruptIfRunning)来取消此任务的执行。

 

 ThreadPoolExecutor详解

 Executor框架最核心的类是ThreadPoolExecutor

 ThreadPoolExecutor的组件构成

• corePool：核心线程池的大小
• maximumPool：最大线程池的大小
• BlockingQueue：用来暂时保存任务的工做队列
• RejectedExecutionHandler：当ThreadPoolExecutor已经关闭或ThreadPoolExecutor已经饱和时(达到了最大线程池的大小且工做队列已满)，execute()方法将要调用的Handler。

 

Executor 可 以 创 建 3 种 类 型 的 ThreadPoolExecutor 线 程 池：

 1. FixedThreadPool

建立固定长度的线程池，每次提交任务建立一个线程，直到达到线程池的最大数量，线程池的大小再也不变化。

这个线程池能够建立固定线程数的线程池。特色就是能够重用固定数量线程的线程池。它的构造源码以下：

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

• FixedThreadPool的corePoolSize和maxiumPoolSize都被设置为建立FixedThreadPool时指定的参数nThreads。
• 0L则表示当线程池中的线程数量操做核心线程的数量时，多余的线程将被当即中止
• 最后一个参数表示FixedThreadPool使用了无界队列LinkedBlockingQueue做为线程池的作工队列，因为是无界的，当线程池的线程数达到corePoolSize后，新任务将在无界队列中等待，所以线程池的线程数量不会超过corePoolSize，同时maxiumPoolSize也就变成了一个无效的参数，而且运行中的线程池并不会拒绝任务。

FixedThreadPool运行图以下

执行过程以下：

1.若是当前工做中的线程数量少于corePool的数量，就建立新的线程来执行任务。

2.当线程池的工做中的线程数量达到了corePool，则将任务加入LinkedBlockingQueue。

3.线程执行完1中的任务后会从队列中去任务。

注意LinkedBlockingQueue是无界队列，因此能够一直添加新任务到线程池。

 

2. SingleThreadExecutor　　

SingleThreadExecutor是使用单个worker线程的Executor。特色是使用单个工做线程执行任务。它的构造源码以下：

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

　　

SingleThreadExecutor的corePoolSize和maxiumPoolSize都被设置1。

其余参数均与FixedThreadPool相同，其运行图以下：

 

执行过程以下：

1.若是当前工做中的线程数量少于corePool的数量，就建立一个新的线程来执行任务。

2.当线程池的工做中的线程数量达到了corePool，则将任务加入LinkedBlockingQueue。

3.线程执行完1中的任务后会从队列中去任务。

注意：因为在线程池中只有一个工做线程，因此任务能够按照添加顺序执行。

 

 3. CachedThreadPool

 CachedThreadPool是一个”无限“容量的线程池，它会根据须要建立新线程。特色是能够根据须要来建立新的线程执行任务，没有特定的corePool。下面是它的构造方法：

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，即maximum是无界的。这里keepAliveTime设置为60秒，意味着空闲的线程最多能够等待任务60秒，不然将被回收。

 

CachedThreadPool使用没有容量的SynchronousQueue做为主线程池的工做队列，它是一个没有容量的阻塞队列。每一个插入操做必须等待另外一个线程的对应移除操做。这意味着，若是主线程提交任务的速度高于线程池中处理任务的速度时，CachedThreadPool会不断建立新线程。极端状况下，CachedThreadPool会由于建立过多线程而耗尽CPU资源。其运行图以下：

 

执行过程以下：

1.首先执行SynchronousQueue.offer(Runnable task)。若是在当前的线程池中有空闲的线程正在执行SynchronousQueue.poll()，那么主线程执行的offer操做与空闲线程执行的poll操做配对成功，主线程把任务交给空闲线程执行。，execute()方法执行成功，不然执行步骤2

2.当线程池为空(初始maximumPool为空)或没有空闲线程时，配对失败，将没有线程执行SynchronousQueue.poll操做。这种状况下，线程池会建立一个新的线程执行任务。

3.在建立完新的线程之后，将会执行poll操做。当步骤2的线程执行完成后，将等待60秒，若是此时主线程提交了一个新任务，那么这个空闲线程将执行新任务，不然被回收。所以长时间不提交任务的CachedThreadPool不会占用系统资源。

SynchronousQueue是一个不存储元素阻塞队列，每次要进行offer操做时必须等待poll操做，不然不能继续添加元素。

 

参考书籍：《Java并发编程的艺术》，《Java并发编程实战》，《Java高并发程序设计》

更多内容：http://www.cnblogs.com/study-everyday/