一、前言
1.5后引入的Executor框架的最大优点是把任务的提交和执行解耦。要执行任务的人只需把Task描述清楚，然后提交即可。这个Task是怎么被执行的，被谁执行的，什么时候执行的，提交的人就不用关心了。具体点讲，提交一个Callable对象给ExecutorService（如最常用的线程池ThreadPoolExecutor），将得到一个Future对象，调用Future对象的get方法等待执行结果就好了。
经过这样的封装，对于使用者来说，提交任务获取结果的过程大大简化，调用者直接从提交的地方就可以等待获取执行结果。而封装最大的效果是使得真正执行任务的线程们变得不为人知。有没有觉得这个场景似曾相识？我们工作中当老大的老大（且称作LD^2）把一个任务交给我们老大（LD）的时候，到底是LD自己干，还是转过身来拉来一帮苦逼的兄弟加班加点干，那LD^2是不管的。LD^2只用把人描述清楚提及给LD，然后喝着咖啡等着收LD的report即可。等LD一封邮件非常优雅地报告LD^2report结果时，实际操作中是码农A和码农B干了一个月，还是码农ABCDE加班干了一个礼拜，大多是不用体现的。这套机制的优点就是LD^2找个合适的LD出来提交任务即可，接口友好有效，不用为具体怎么干费神费力。
二、 一个最简单的例子
看上去这个执行过程是这个样子。调用这段代码的是老大的老大了，他所需要干的所有事情就是找到一个合适的老大（如下面例子中laodaA就荣幸地被选中了），提交任务就好了。
// 一个有7个作业线程的线程池，老大的老大找到一个管7个人的小团队的老大
ExecutorService laodaA = Executors.newFixedThreadPool(7);
//提交作业给老大，作业内容封装在Callable中，约定好了输出的类型是String。
String outputs = laoda.submit(
new Callable<String>() {
public String call() throws Exception
{
return "I am a task, which submited by the so called laoda, and run by those anonymous workers";
}
//提交后就等着结果吧，到底是手下7个作业中谁领到任务了，老大是不关心的。
}).get();
System.out.println(outputs);
使用上非常简单，其实只有两行语句来完成所有功能：创建一个线程池，提交任务并等待获取执行结果。
例子中生成线程池采用了工具类Executors的静态方法。除了newFixedThreadPool可以生成固定大小的线程池，newCachedThreadPool可以生成一个无界、可以自动回收的线程池，newSingleThreadScheduledExecutor可以生成一个单个线程的线程池。newScheduledThreadPool还可以生成支持周期任务的线程池。一般用户场景下各种不同设置要求的线程池都可以这样生成，不用自己new一个线程池出来。
三、代码剖析
这套机制怎么用，上面两句语句就做到了，非常方便和友好。但是submit的task是怎么被执行的？是谁执行的？如何做到在调用的时候只有等待执行结束才能get到结果。这些都是1.5之后Executor接口下的线程池、Future接口下的可获得执行结果的的任务，配合AQS和原有的Runnable来做到的。在下文中我们尝试通过剖析每部分的代码来了解Task提交，Task执行，获取Task执行结果等几个主要步骤。为了控制篇幅，突出主要逻辑，文章中引用的代码片段去掉了异常捕获、非主要条件判断、非主要操作。文中只是以最常用的ThreadPoolExecutor线程池举例，其实ExecutorService接口下定义了很多功能丰富的其他类型，有各自的特点，但风格类似。本文重点是介绍任务提交的过程，过程中涉及的ExecutorService、ThreadPoolExecutor、AQS、Future、FutureTask等只会介绍该过程中用到的内容，不会对每个类都详细展开。
1、 任务提交
从类图上可以看到，接口ExecutorService继承自Executor。不像Executor中只定义了一个方法来执行任务，在ExecutorService中，正如其名字暗示的一样，定义了一个服务，定义了完整的线程池的行为，可以接受提交任务、执行任务、关闭服务。抽象类AbstractExecutorService类实现了ExecutorService接口，也实现了接口定义的默认行为。
(点击放大图像)
AbstractExecutorService任务提交的submit方法有三个实现。第一个接收一个Runnable的Task，没有执行结果；第二个是两个参数：一个任务，一个执行结果；第三个一个Callable，本身就包含执任务内容和执行结果。 submit方法的返回结果是Future类型，调用该接口定义的get方法即可获得执行结果。 V get() 方法的返回值类型V是在提交任务时就约定好了的。
除了submit任务的方法外，作为对服务的管理，在ExecutorService接口中还定义了服务的关闭方法shutdown和shutdownNow方法，可以平缓或者立即关闭执行服务，实现该方法的子类根据自身特征支持该定义。在ThreadPoolExecutor中，维护了RUNNING、SHUTDOWN、STOP、TERMINATED四种状态来实现对线程池的管理。线程池的完整运行机制不是本文的重点，重点还是关注submit过程中的逻辑。
1) 看AbstractExecutorService中代码提交部分，构造好一个FutureTask对象后，调用execute()方法执行任务。我们知道这个方法是顶级接口Executor中定义的最重要的方法。。FutureTask类型实现了Runnable接口，因此满足Executor中execute()方法的约定。同时比较有意思的是，该对象在execute执行后，就又作为submit方法的返回值返回，因为FutureTask同时又实现了Future接口，满足Future接口的约定。
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
2) Submit传入的参数都被封装成了FutureTask类型来execute的，对应前面三个不同的参数类型都会封装成FutureTask。
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
3) Executor接口中定义的execute方法的作用就是执行提交的任务，该方法在抽象类AbstractExecutorService中没有实现，留到子类中实现。我们观察下子类ThreadPoolExecutor，使用最广泛的线程池如何来execute那些submit的任务的。这个方法看着比较简单，但是线程池什么时候创建新的作业线程来处理任务，什么时候只接收任务不创建作业线程，另外什么时候拒绝任务。线程池的接收任务、维护工作线程的策略都要在其中体现。
作为必要的预备知识，先补充下ThreadPoolExecutor有两个最重要的集合属性，分别是存储接收任务的任务队列和用来干活的作业集合。
//任务队列 private final BlockingQueue<Runnable> workQueue; //作业线程集合 private final HashSet<Worker> workers = new HashSet<Worker>();
其中阻塞队列workQueue是来存储待执行的任务的，在构造线程池时可以选择满足该BlockingQueue 接口定义的SynchronousQueue、LinkedBlockingQueue或者DelayedWorkQueue等不同阻塞队列来实现不同特征的线程池。
关注下execute(Runnable command)方法中调用到的addIfUnderCorePoolSize，workQueue.offer(command) ， ensureQueuedTaskHandled(command)，addIfUnderMaximumPoolSize(command)这几个操作。尤其几个名字较长的private方法，把方法名的驼峰式的单词分开，加上对方法上下文的了解就能理解其功能。
因为前面说到的几个方法在里面即是操作，又返回一个布尔值，影响后面的逻辑，所以不大方便在方法体中为每条语句加注释来说明，需要大致关联起来看。所以首先需要把execute方法的主要逻辑说明下，再看其中各自方法的作用。
- 如果线程池的状态是RUNNING，线程池的大小小于配置的核心线程数，说明还可以创建新线程，则启动新的线程执行这个任务。
- 如果线程池的状态是RUNNING ，线程池的大小小于配置的最大线程数，并且任务队列已经满了，说明现有线程已经不能支持当前的任务了，并且线程池还有继续扩充的空间，就可以创建一个新的线程来处理提交的任务。
- 如果线程池的状态是RUNNING，当前线程池的大小大于等于配置的核心线程数，说明根据配置当前的线程数已经够用，不用创建新线程，只需把任务加入任务队列即可。如果任务队列不满，则提交的任务在任务队列中等待处理；如果任务队列满了则需要考虑是否要扩展线程池的容量。
- 当线程池已经关闭或者上面的条件都不能满足时，则进行拒绝策略，拒绝策略在RejectedExecutionHandler接口中定义，可以有多种不同的实现。
上面其实也是对最主要思路的解析，详细展开可能还会更复杂。简单梳理下思路：构建线程池时定义了一个额定大小，当线程池内工作线程数小于额定大小，有新任务进来就创建新工作线程，如果超过该阈值，则一般就不创建了，只是把接收任务加到任务队列里面。但是如果任务队列里的任务实在太多了，那还是要申请额外的工作线程来帮忙。如果还是不够用就拒绝服务。这个场景其实也是每天我们工作中会碰到的场景。我们管人的老大，手里都有一定HC（Head Count），当上面老大有活分下来，手里人不够，但是不超过HC，我们就自己招人；如果超过了还是忙不过来，那就向上门老大申请借调人手来帮忙；如果还是干不完，那就没办法了，新任务咱就不接了。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}
4) addIfUnderCorePoolSize方法检查如果当前线程池的大小小于配置的核心线程数，说明还可以创建新线程，则启动新的线程执行这个任务。
private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
//如果当前线程池的大小小于配置的核心线程数，说明还可以创建新线程
if (poolSize < corePoolSize && runState == RUNNING)
// 则启动新的线程执行这个任务
t = addThread(firstTask);
return t != null;
}
5) 和上一个方法类似，addIfUnderMaximumPoolSize检查如果线程池的大小小于配置的最大线程数，并且任务队列已经满了（就是execute方法试图把当前线程加入任务队列时不成功），说明现有线程已经不能支持当前的任务了，但线程池还有继续扩充的空间，就可以创建一个新的线程来处理提交的任务。
private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
// 如果线程池的大小小于配置的最大线程数，并且任务队列已经满了（就
是execute方法中试图把当前线程加入任务队列workQueue.offer(command)时候不成功
）,说明现有线程已经不能支持当前的任务了，但线程池还有继续扩充的空间
if (poolSize < maximumPoolSize && runState == RUNNING)
//就可以创建一个新的线程来处理提交的任务
t = addThread(firstTask);
return t != null;
}
6) 在ensureQueuedTaskHandled方法中，判断如果当前状态不是RUNING，则当前任务不加入到任务队列中，判断如果状态是停止，线程数小于允许的最大数，且任务队列还不空，则加入一个新的工作线程到线程池来帮助处理还未处理完的任务。
private void ensureQueuedTaskHandled(Runnable command) {
// 如果当前状态不是RUNING，则当前任务不加入到任务队列中，判断如
果状态是停止，线程数小于允许的最大数，且任务队列还不空
if (state < STOP &&
poolSize < Math.max(corePoolSize, 1) &&
!workQueue.isEmpty())
//则加入一个新的工作线程到线程池来帮助处理还未处理完的任务
t = addThread(null);
if (reject)
reject(command);
}
7) 在前面方法中都会调用adThread方法创建一个工作线程，差别是创建的有些工作线程上面关联接收到的任务firstTask，有些没有。该方法为当前接收到的任务firstTask创建Worker，并将Worker添加到作业集合HashSet<Worker> workers中，并启动作业。
private Thread addThread(Runnable firstTask) {
//为当前接收到的任务firstTask创建Worker
Worker w = new Worker(firstTask);
Thread t = threadFactory.newThread(w);
w.thread = t;
//将Worker添加到作业集合HashSet<Worker> workers中，并启动作业
workers.add(w);
t.start();
return t;
}
至此，任务提交过程简单描述完毕，并介绍了任务提交后ExecutorService框架下线程池的主要应对逻辑，其实就是接收任务，根据需要创建或者维护管理线程。
维护这些工作线程干什么用？先不用看后面的代码，想想我们老大每月辛苦地把老板丰厚的薪水递到我们手里，定期还要领着大家出去happy下，又是定期的关心下个人生活，所有做的这些都是为什么呢？木讷的代码工不往这边使劲动脑子，但是猜还是能猜的到的，就让干活呗。本文想着重表达细节，诸如线程池里的Worker是怎么工作的，Task到底是不是在这些工作线程中执行的，如何保证执行完成后，外面等待任务的老大拿到想要结果，我们将在下篇文章中详细介绍。
上一篇文章中通过引入的一个例子介绍了在Executor框架下，提交一个任务的过程，这个过程就像我们老大的老大要找个老大来执行一个任务那样简单。并通过剖析ExecutorService的一种经典实现ThreadPoolExecutor来分析接收任务的主要逻辑，发现ThreadPoolExecutor的工作思路和我们带项目的老大的工作思路完全一致。在本文中我们将继续后面的步骤，着重描述下任务执行的过程和任务执行结果获取的过程。会很容易发现，这个过程我们更加熟悉，因为正是每天我们工作的过程。除了ThreadPoolExecutor的内部类Worker外，对执行内容和执行结果封装的FutureTask的表现是这部分着重需要了解的。
为了连贯期间，内容的编号延续上篇。
2. 任务执行
其实应该说是任务被执行，任务是宾语。动宾结构：execute the task，执行任务，无论写成英文还是中文似乎都是这样。那么主语是是who呢？明显不是调用submit的那位（线程），那是哪位呢？上篇介绍ThreadPoolExecutor主要属性时提到其中有一个HashSet<Worker> workers的集合，我们有说明这里存储的就是线程池的工作队列的集合，队列的对象是Worker类型的工作线程，是ThreadPoolExecutor的一个内部类，实现了Runnable接口：
private final class Worker implements Runnable
8) 看作业线程干什么当然是看它的run方法在干什么。如我们所料，作业线程就是在一直调用getTask方法获取任务，然后调用 runTask(task)方法执行任务。看到没有，是在while循环里面，就是不干完不罢休的意思！在加班干活的苦逼的朋友们，有没有遇见战友的亲切感觉？
然后简单看下getTask和runTask(task)方法的内容。
9) getTask方法是ThreadPoolExecutor提供给其内部类Worker的的方法。作用就是一个，从任务队列中取任务，源源不断地输出任务。有没有想到老大手里拿的总是满满当当的project，也是源源不断的。
Runnable getTask() {
for (;;) {
//从任务队列的头部取任务
r = workQueue.take();
return r;
}
}
10) runTask(Runnable task)是工作线程Worker真正处理拿到的每个具体任务。看到这里才可用确认我们的猜想，之前提到[y1] 的“执行任务”这个动宾结构前面的主语正是这些Worker呀。唠叨了半天（看主要方法都看到了整整第10个了），前面都是派活，这里才是干活。和我们的工作何其相似！老大（LD），老大的老大（LD^2），老大的老大（LD^n） 非常辛苦，花了很多时间、精力在会议室、在project上想着怎么生成和安排任务，然而真的轮到咱哥们干活，可能花了不少时间，但看看流程就是这么简单。三个大字：“Just do it”。
private void runTask(Runnable task) {
//调用任务的run方法，即在Worker线程中执行Task内定义内容。
task.run();
}
需要注意的地方出现了，调用的其实是task的run方法。看下FutureTask的run方法做了什么事情。
这里插入一个FutureTask的类图。可以看到FutureTask实现了RunnableFuture接口，所以FutureTask即有Runnable接口的run方法来定义任务内容，也有Future接口中定义的get、cancel等方法来控制任务执行和获取执行结果。Runnable接口自不用说，Future接口的伟大设计，就是使得实现该接口的对象可以阻塞线程直到任务执行完毕，也可以取消任务执行，检测任务是执行完毕还是被取消了。想想在之前我们使用Thread.join()或者Thread.join(long millis)等待任务结束是多么苦涩啊。
FutureTask内部定义了一个Sync的内部类，继承自AQS，来维护任务状态。关于AQS的设计思路，可以参照参考Doug Lea大师的原著The java.util.concurrent Synchronizer Framework。
(点击放大图像)
11) 和其他的同步工具类一样，FutureTask的主要工作内容也是委托给其定义的内部类Sync来完成。
public void run() {
//调用Sync的对应方法
sync.innerRun();
}
12) FutureTask.Sync.innerRun()，这样做的目的就是为了维护任务执行的状态，只有当执行完后才能够获得任务执行结果。在该方法中，首先设置执行状态为RUNNING只有判断任务的状态是运行状态，才调用任务内封装的回调，并且在执行完成后设置回调的返回值到FutureTask的result变量上。在FutureTask中，innerRun等每个“写”方法都会首先修改状态位，在后续会看到innerGet等“读”方法会先判断状态，然后才能决定后续的操作是否可以继续。下图是FutureTask.Sync中几个重要状态的流转情况，和其他的同步工具类一样，状态位使用的也是父类AQS的state属性。
(点击放大图像)
void innerRun() {
//通过对AQS的状态位state的判断来判断任务的状态是运行状态，则调用任务内封装的回调，并且设置回调的返回值
if (getState() == RUNNING)
innerSet(callable.call());
}
void innerSet(V v) {
for (;;) {
int s = getState();
//设置运行状态为完成，并且把回调额执行结果设置给result变量
if (compareAndSetState(s, RAN)) {
result = v;
releaseShared(0);
done();
return;
}
}
至此工作线程执行Task就结束了。提交的任务是由Worker工作线程执行，正是在该线程上调用Task中定义的任务内容，即封装的Callable回调，并设置执行结果。下面就是最重要的部分：调用者如何获取执行的结果。让你加班那么久，总得把成果交出来吧。老大在等，因为老大的老大在等！
3. 获取执行结果
前面说过，对于老大的老大这样的使用者来说，获取执行结果这个过程总是最容易的事情，只需调用FutureTask的get()方法即可。该方法是在Future接口中就定义的。get方法的作用就是等待执行结果。（Waits if necessary for the computation to complete, and then retrieves its result.）Future这个接口命名得真好，虽然是在未来，但是定义有一个get()方法，总是“可以掌控的未来，总是有收获的未来！”实现该接口的FutureTask也应该是这个意思，在未来要完成的任务，但是一样要有结果哦。
13) FutureTask的get方法同样委托给Sync来执行。和该方法类似，还有一个V get(long timeout, TimeUnit unit)，可以配置超时时间。
public V get() throws InterruptedException, ExecutionException {
return sync.innerGet();
}
14) 在Sync的 innerGet方法中，调用AQS父类定义的获取共享锁的方法acquireSharedInterruptibly来等待执行完成。如果执行完成了则可以继续执行后面的代码，返回result结果，否则如果还未完成，则阻塞线程等待执行完成。[bd2] 再大的老大要想获得结果也得等老子干完了才行！可以看到调用FutureTask的get方法，进而调用到该方法的一定是想要执行结果的线程，一般应该就是提交Task的线程，而这个任务的执行是在Worker的工作线程上，通过AQS来保证执行完毕才能获取执行结果。该方法中acquireSharedInterruptibly是AQS父类中定义的获取共享锁的方法，但是到底满足什么条件可以成功获取共享锁，这是Sync的tryAcquireShared方法内定义的。[bd3] 具体说来，innerIsDone用来判断是否执行完毕，如果执行完毕则向下执行，返回result即可；如果判断未完成，则调用AQS的doAcquireSharedInterruptibly来挂起当前线程，一直到满足条件。这种思路在其他的几种同步工具类Semaphore、CountDownLatch、ReentrantLock、ReentrantReadWriteLock也广泛使用。借助AQS框架，在获取锁时，先判断当前状态是否允许获取锁，若是允许则获取锁，否则获取不成功。获取不成功则会阻塞，进入阻塞队列。而释放锁时，一般会修改状态位，唤醒队列中的阻塞线程。每个同步工具类的自定义同步器都继承自AQS父类，是否可以获取锁根据同步类自身的功能要求覆盖AQS对应的try前缀方法，这些方法在AQS父类中都是只有定义没有内容。可以参照《源码剖析AQS在几个同步工具类中的使用》来详细了解。
突然想到想想那些被称为老大的，是不是整个career流程就是只干两件事情：submit a task， then wait and get the result。不对，还有一件事情，不是等待，而是催。“完了没，完了没？schedule很紧的，抓点紧啊，要不要适当加点班啊……”
V innerGet() throws InterruptedException, ExecutionException {
//获得锁，表示执行完毕，才能获得后执行结果，否则阻塞等待执行完成再获取执行结果
acquireSharedInterruptibly(0);
return result;
}
protected int tryAcquireShared(int ignore) {
return innerIsDone()? 1 : -1;
}
至此，获得执行结果，圆满完成任务！
老大的老大，拍着咱们老大的肩膀（或者深情的抚摸着咱们老大唏嘘胡茬的脸庞）说：“亲，你这活干的漂亮！”而隔壁桌座位的几个兄弟，刚熬了几个晚上加班交付完这波task后，发现任务队列里又有新任务了，俺们老大又从他的另外一个老大手里接来的任务了。每个人都按照这样的角色进行着，依照这样的角色安排和谐愉快地进行着。。。
|
角色名 |
任务用户 |
任务管理者 |
任务执行者 |
| 角色属性 | 任务的甲方 | 任务的乙方 | 乙方的工具 |
| 角色说明 | 选择合适的任务执行服务，如可以根据需要选择ThreadPoolExecutor还是ScheduledThreadPoolExecutor，并定制ExecutorService的配置。 定义好任务的工作内容和结果类型，提交任务，等待任务的执行结果 |
接收提交的任务； 维护执行服务内部管理； 配置工作线程执行任务 |
每个工作线程一直从任务执行服务获取待执行的任务，保证任务完成后返回执行结果。 |
| Executor中对应 | 创建获取ExecutorService、并提交Task的外部接口 | ExecutorService的各种实现。如经典的ThreadPoolExecutor，ScheduledThreadPoolExecutor | 执行服务内定义的配套的Worker线程。如ThreadPoolExecutor.Worker |
| 主要接口方法 | submit(Callable task) | execute(Runnable command) | runTask(Runnable task) |
| 现实角色映射 | 手里有活的大老大 | 领人干活的老大 | 真正干活的码农 |
| 主要工作伪代码 | taskService = createService() future=taskService.submitTask() future.get() |
executeTask() { addTask() createThread() } |
while(ture) { getTask() runTask() } |
四、 总结
从时序图上看主要的几个角色是这样配合完成任务提交、任务执行、获取执行结果这几个步骤的。
(点击放大图像)
- 外面需要提交任务的角色（如例子中老大的老大），首先创建一个任务执行服务ExecutorService，一般使用工具类Executors的若干个工厂方法 创建不同特征的线程池ThreadPoolExecutor，例子中是使用newFixedThreadPool方法创建有n个固定工作线程的线程池。
- 线程池是专门负责从外面接活的老大。把任务封装成一个FutureTask对象，并根据输入定义好要获得结果的类型，就可以submit任务了。
- 线程池就像我们团队里管人管项目的老大，各个都有一套娴熟、有效的办法来对付输入的任务和手下干活的兄弟一样，内部有一套比较完整、细致的任务管理办法，工作线程管理办法，以便应付输入的任务。这些逻辑全部在其execute方法中体现。
- 线程池接收输入的task，根据需要创建工作线程，启动工作线程来执行task。
- 工作线程在其run方法中一直循环，从线程池领取可以执行的task，调用task的run方法执行task内定义的任务。
- FutureTask的run方法中调用其内部类Sync的innerRun方法来执行封装的具体任务，并把任务的执行结果返回给FutureTask的result变量。
- 当提及任务的角色调用FutureTask的get方法获取执行结果时，Sync的innerGet方法被调用。根据任务的执行状态判断，任务执行完毕则返回执行结果；未执行完毕则等待。
还记得我们费了半天劲试图找出任务执行时那个动宾结构的主语吗？从示例上看更像是线程池在向外提供任务执行的服务。就像我们的老大在代表我们接收任务、执行任务、提交执行结果。明显我们这些真正的Worker成了延伸，有点搞不懂到底我们是主语，还是主语延伸的工具，就像定义ThreadPoolExecutor的内部类Worker一样。我们只是工具，不是主语，是状语： execute the task by workers。突然想到毛主席当年的“数风流人物，还看今朝”，说的应该是这些Worker的劳苦大众吧，怎么都今朝这么久了，俺们这些Woker们还是风流不起来呢？风骚的作者居然在上面严肃的时序图上加了个风骚的小星星，向同行的Worker们致敬！