Executor框架的使用简介

类似于我们熟悉的集合框架（由Collection和Map接口衍生出很多其他的接口和类），在JAVA多线程中，也存在一个Executor框架。等以后时间充足了，会对该框架来一波源码剖析。

简而言之，Executor框架实现了工作单元与执行单元的分离。

本文用到的程序源码请参考我的github。

一.Executor框架的两级调度模型

在HotSpot VM的线程模型中，JAVA线程被一对一映射为本地操作系统线程。JAVA线程启动时会启动一个本地操作系统线程：当该JAVA线程终止时，这个操作系统线程也会被回收。操作系统会调度所有线程并将它们分配给可用的CPU。

两级调度模型的示意图：



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

二.Executor框架的结构

Executor主要由三部分组成：任务产生部分，任务处理部分，结果获取部分。（设计模式：生产者与消费者模式）

先来看个图：

1.任务的产生：Runnable接口和Callable接口

这2个对象属于任务对象。工具类Executors可以把一个Runnable对象封装为Callable对象。当我们拥有任务对象之后，就可以将其交给ExecutorService（Executor的一个实现接口）了，这样转入第二部分–任务处理部分。

2.任务的处理：Executor接口—>ExecutorService接口

任务的处理主要是将任务丢到线程池中，由线程池提供线程将任务“消费掉”。

线程池有2类：ThreadPoolExecutor和ScheduledThreadPoolExecutor。2种线程池类均可以通过工厂类Executors来创建。

⑴：ThreadPoolExecutor类

工厂类可以创建3种类型的ThreadPoolExecutor类：

①：FixedThreadPool：拥有固定数量线程的线程池，限制了线程的数目，适用于负载比较重的服务器。

②：SingleThreadPool：单个线程的线程池，适用于需要保证顺序的执行各个任务；任意时间点，不会有多个线程活动。

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

⑵：ScheduleThreadPoolExecutor类

工厂类可以创建2种类型的SchedulePoolExecutor类：

①：ScheduleThreadPoolExecutor：包含若干线程。

②：SingleThreadScheduleExecutor:单个线程。

3.任务结果的获取：Future接口

Future接口有个实现类FutureTask，迄今为止API中返回的都是FutureTask对象，未来的JDK实现中，可能有Future对象。

关于Executor框架的架构基本就这些，下面用几个Demo实测一下：

1.下面这个程序实现了启动一个2个线程的线程池并给该线程池扔入5个任务，结果显示了他们的执行策略：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
* @author Vayne
*
* Executors是线程池框架的一个工具类
*
*
*/
public class FixedThreadPoolDemo
{
public static void main(String[] args)
{
ExecutorService pool = Executors.newFixedThreadPool(2);

// 定义一个循环，添加5个任务
for (int i = 0; i < 5; i++)
{
int flag = i;
pool.execute(new Runnable()
{
//任务详情：执行6次打印语句。
@Override
public void run()
{
for(int j = 0;j<6;j++)
{
try
{
Thread.sleep(10);
} catch (InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+" "+flag+" "+" `s loop : "+j);
}
}
});
}

pool.shutdown();

}
}

结果：

pool-1-thread-1 0  `s loop : 0
pool-1-thread-2 1  `s loop : 0
pool-1-thread-1 0  `s loop : 1
pool-1-thread-2 1  `s loop : 1
pool-1-thread-1 0  `s loop : 2
pool-1-thread-2 1  `s loop : 2
pool-1-thread-2 1  `s loop : 3
pool-1-thread-1 0  `s loop : 3
pool-1-thread-1 0  `s loop : 4
pool-1-thread-2 1  `s loop : 4
pool-1-thread-1 0  `s loop : 5
pool-1-thread-2 1  `s loop : 5           //这里2个线程将2个任务完成
pool-1-thread-1 3  `s loop : 0
pool-1-thread-2 2  `s loop : 0
pool-1-thread-2 2  `s loop : 1
pool-1-thread-1 3  `s loop : 1
pool-1-thread-2 2  `s loop : 2
pool-1-thread-1 3  `s loop : 2
pool-1-thread-2 2  `s loop : 3
pool-1-thread-1 3  `s loop : 3
pool-1-thread-2 2  `s loop : 4
pool-1-thread-1 3  `s loop : 4
pool-1-thread-1 3  `s loop : 5
pool-1-thread-2 2  `s loop : 5          //2个任务完成
pool-1-thread-1 4  `s loop : 0
pool-1-thread-1 4  `s loop : 1
pool-1-thread-1 4  `s loop : 2
pool-1-thread-1 4  `s loop : 3
pool-1-thread-1 4  `s loop : 4
pool-1-thread-1 4  `s loop : 5          //最后1个任务完成

下面我们再来看看CachedThreadPool的实测结果，由于代码区别只是构造方法的不同，这里不浪费页面贴代码了，直接贴出结果：

pool-1-thread-2 1  `s loop : 0
pool-1-thread-4 3  `s loop : 0
pool-1-thread-5 4  `s loop : 0
pool-1-thread-3 2  `s loop : 0
pool-1-thread-1 0  `s loop : 0    //一次开启了5个线程来承载5次任务
pool-1-thread-4 3  `s loop : 1
pool-1-thread-2 1  `s loop : 1
pool-1-thread-5 4  `s loop : 1
pool-1-thread-3 2  `s loop : 1
pool-1-thread-1 0  `s loop : 1
pool-1-thread-4 3  `s loop : 2
pool-1-thread-2 1  `s loop : 2
pool-1-thread-1 0  `s loop : 2
pool-1-thread-5 4  `s loop : 2
pool-1-thread-3 2  `s loop : 2
pool-1-thread-2 1  `s loop : 3
pool-1-thread-4 3  `s loop : 3
pool-1-thread-1 0  `s loop : 3
pool-1-thread-5 4  `s loop : 3
pool-1-thread-3 2  `s loop : 3
pool-1-thread-2 1  `s loop : 4
pool-1-thread-4 3  `s loop : 4
pool-1-thread-1 0  `s loop : 4
pool-1-thread-5 4  `s loop : 4
pool-1-thread-3 2  `s loop : 4
pool-1-thread-2 1  `s loop : 5
pool-1-thread-4 3  `s loop : 5
pool-1-thread-5 4  `s loop : 5
pool-1-thread-1 0  `s loop : 5
pool-1-thread-3 2  `s loop : 5

2.结合Future接口来做一个实测

import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FutureDemo
{
public static void main(String[] args)
{

ExecutorService pool = Executors.newFixedThreadPool(3);

//CompletionService接口内部维护一个结果队列:一堆future....
CompletionService<Integer> cs = new ExecutorCompletionService<>(pool);

for (int i = 1; i < 11; i++)
{
final int flag = i * 10;
cs.submit(new Callable<Integer>()
{

@Override
public Integer call() throws Exception
{
// TODO Auto-generated method stub
Thread.sleep(1000);
return flag;
}
});
}

for (int i = 0; i < 11; i++)
{
try
{
System.out.println(cs.take().get());
} catch (InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ExecutionException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}

pool.shutdown();

}
}

结果：

10
20
30
50
40
60
70
90
80
100