1 内容回顾
前面四节的内容我们大概看了线程池的：
（1）线程池的基本用法
（2）线程池种类ExecuteService这类型的线程池，代表的子类是ThreadPoolExecutor，这种类型的线程池是当有线程空闲的时候立即会执行你提交的任务。还有一种类型的线程池ScheduledExecutorService 这种类型的线程池是调度类型的，允许你提交延时任务、定时任务、延时定时任务等。
（3）讲解了ThreadPoolExecutor线程池内部的各种变量、核心参数的意思、线程拒绝策略等
（4）讲解了ThreadPoolExecutor提供的execute方法，提交任务的方法内部流程是怎么样的
（5）讲解了Worker内部的工作原理，怎么不断运行提交到线程池的任务的，空闲时候怎么被销毁的
（6）讲解了ThreadPoolExecutor线程池的预热、关闭、线程和任务的统计类方法
那么我们本节要看的是线程池提交任务的另外一个方法submit方法以及submit方法关联的Future、FutureTask等原理。
2 submit 方法
我们来先看看submit方法内部有什么逻辑，ThreadPoolExecutor继承了AbstractExecutorService，submit方法就在这个抽象线程池里面：
// 这里传入一个Callable<T> task任务 // submit有几个重载方法，传入的无论是Callable、Runnable，内部的实现流程都是一样的 public <T> Future<T> submit(Callable<T> task) { // 校验一下传入的任务，如果是空，则抛出异常 if (task == null) throw new NullPointerException(); // 将传入的task任务包装成一个RunnableFuture RunnableFuture<T> ftask = newTaskFor(task); // 这里就是调用线程池内部的execute方法来执行类，这里的execute方法我们之前分析过了 execute(ftask); // 这里就将包装得到的RunnableTask返回 return ftask; }
我们继续来看下newTaskFor方法是怎么对task任务进行包装的：
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { // 很简单，就是创建出来一个FutureTask实例，然后将callable任务封装在这个FutureTask中 return new FutureTask<T>(callable); }
上面的submit任务方法流程很简单，我们画个图理解一下：
如上图所示：
（1）submit方法的核心其实就是，将传入的task任务封装成一个FutureTask。
（2）然后将这个封装好的FutureTask（实现了runnable接口）交给execute方法提交任务，execute方法内部逻辑我们之前详细分析过了
（3）然后就是将这个FutureTask返回给调用者就完事了
（4）所以submit和execute不同点在于构建了一个FutureTask对象，FutureTask这里就是我们接下来分析的重点了。
3 Future 使用
在看FutureTask之前我们先写个例子，回顾下我们平时使用的Future，Future整合线程池的submit方法我们平时业务经常使用，比如：
public class FutureTaskTest { /** * 创建一个线程池 */ public static ThreadPoolExecutor executor = new ThreadPoolExecutor( 2, 5, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new DefaultThreadFactory("future-pool-test"), new ThreadPoolExecutor.CallerRunsPolicy() ); public static void main(String[] args) throws ExecutionException, InterruptedException { // 封装一个Runnable任务 Runnable runnableTask = () -> { System.out.println("【线程池线程】::当前时间:"+new Date() + " 开始执行【Runnable】任务"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("【线程池线程】::当前时间:"+new Date() + " 开始执行【Runnable】任务"); }; // 将一个runnable任务通过submit方法提交给线程池 // 线程池返回一个Future接口的实现类 // 上面我们分析过了，其实就是返回一个FutureTask实例 Future future = executor.submit(runnableTask); // 这里主线程调用future的get接口会被阻塞住，知道runnable任务执行成功为止 // 这里获取任务的结果，由于是Runnable接口没有返回值，所以得到结果一定是Null Object result = future.get(); System.out.println("【主线程/调用线程】当前时间:"+new Date() + " 执行【runnableTask】结束, 获取结果为:" + result); } }
主线程的打印时间是在Runnable接口日志打印之后，说明调用Future的get接口阻塞了主线程，直到Runnable接口成功为止。
把上面的Runnable接口换成有返回值的Callable接口如下：
public static void main(String[] args) throws ExecutionException, InterruptedException { // 封装一个Callable任务 Callable<Integer> callTask = () -> { System.out.println("【线程池线程】::当前时间:"+new Date() + " 开始执行【Callable】任务"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("【线程池线程】::当前时间:"+new Date() + " 开始执行【Callable】任务"); // 执行结束后返回结果值 return 10000; }; // 将一个Callable任务通过submit方法提交给线程池 // 线程池返回一个Future接口的实现类 // 上面我们分析过了，其实就是返回一个FutureTask实例 Future<Integer> future = executor.submit(callTask); // 这里主线程调用future的get接口会被阻塞住，知道Callable任务执行成功为止 // 这里获取任务的结果，根据返回值应该得到结果是 10000 Integer result = future.get(); System.out.println("【主线程/调用线程】当前时间:"+new Date() + " 执行【callableTask】结束, 获取结果为:" + result); }
这里也是跟上面的例子一样：
（1）调用Future接口的get方法的线程会被阻塞住，直到线程池执行完任务之后将结果值返回
（2）这里使用Callable接口，跟Runnable不一样的是，Callable接口是有返回值的
（3）很多情况下，可以通过这种Callable、Future、submit这种方式交给线程池执行，但是又可以同步获取结果
那我们接下来就来看看Future接口的get方法是怎么实现让主线程阻塞，知道Runnable任务或者Callable任务执行成功的
4 FutureTask 源码
阻塞机制就是封装在FutureTask对象里面的，我们看一下FutureTask这个对象内部是怎么实现的：
4.1 接口及内部属性
我们首先看下RunnableFuture接口，继承了Runnable、Future这两个接口，再看下FutureTask这个类实现了RunnableFuture接口，及其内部的属性：
public interface RunnableFuture<V> extends Runnable, Future<V> { void run(); }
public class FutureTask<V> implements RunnableFuture<V> { // 要执行的任务 private Callable<V> callable; // Callable任务的返回结果 private Object outcome; // 运行任务的线程 private volatile Thread runner; // 内部有一个state变量，表示任务执行的状态 private volatile int state; // 表示FutureTask刚创建出来 private static final int NEW = 0; // 完成中 private static final int COMPLETING = 1; // 已完成状态 private static final int NORMAL = 2; // 异常，运行过程抛出异常 private static final int EXCEPTIONAL = 3; // 已经被取消 private static final int CANCELLED = 4; // 中断进行中 private static final int INTERRUPTING = 5; // 已经被中断 private static final int INTERRUPTED = 6; // 阻塞线程列表（由于调用Future.get方法而被阻塞的线程链表） // 任务完成之后，会把该链表的线程一个个唤醒 private volatile WaitNode waiters; }
4.2 构造方法
public FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); // 保存一下callable任务 this.callable = callable; // 内部任务的状态初始化为NEW(刚创建) this.state = NEW; }
public FutureTask(Runnable runnable, V result) { // 这里将Runnable任务进行包装成一个Callable任务并保存起来 this.callable = Executors.callable(runnable, result); // 任务的状态为NEW（刚创建） this.state = NEW; }
上面这些方法只是一些FutureTask的变量，构造方法等，最重要的还是state（任务的状态）、outcome（任务的结果）这些东西。
4.3 get 方法
下面继续看FutureTask的get方法，看看它内部到底是通过什么方式来阻塞调用线程的：
public V get() throws InterruptedException, ExecutionException { // 获取当前任务的执行状态 int s = state; // 如果状态小于等于COMPLETING，说明任务还未完成 // 如果任务完成，那么状态应该是NORMAL状态 // 所以当前线程应该阻塞，直到任务完成被唤醒 if (s <= COMPLETING) // 这里就是实际阻塞调用者线程的方法 s = awaitDone(false, 0L); // 走到这里说明任务已经完成，调用report方法返回任务的执行结果 return report(s); }
4.3.1 awaitDone 方法
核心的逻辑就在awaitDone方法中，我们继续awaitDown方法源码：
// 这里timed表示是否有时间限制的阻塞 // false：没有超时时间限制，会一直阻塞直到任务完成 // true：有超时时间限制，会阻塞nacos的时间，如果超时了就会唤醒调用线程 private int awaitDone(boolean timed, long nanos) throws InterruptedException { // 获取超时时间，也就是截止时间 final long deadline = timed ? System.nanoTime() + nanos : 0L; // 初始化一个等待节点（当前线程被封装到一个等待节点中，进入等待链表阻塞等待） WaitNode q = null; boolean queued = false; // for循环重试，直到操作完成 for (;;) { // 如果当前被中断了，那么不需要阻塞了 if (Thread.interrupted()) { // 被中断线程从等待链表中移除 removeWaiter(q); // 抛出中断异常 throw new InterruptedException(); } // 获取当前任务执行状态 int s = state; // 如果任务状态大于COMPLETING说明任务已完成、被取消、被中断等，不需要再阻塞等待 // 只有小于NORMAL的任务才需要被阻塞 if (s > COMPLETING) { if (q != null) q.thread = null; return s; } // 如果任务状态是完成中..。,说明很快就会完成了，先让出cpu一段时间，也不需要进行阻塞了 else if (s == COMPLETING) // 这里就相当于让出cpu一下 Thread.yield(); // 走到这里state状态肯定是NEW，说明任务刚状态或者是运行中 // 肯定是主要阻塞调用线程的 else if (q == null) // 如果q为null，初始化一个等待节点 q = new WaitNode(); // 走到这里q不为null，并且queued为false表示还没入队成功 else if (!queued) // 这里的queued表示是否入队成功，cas操作设置当前waitNode等待节点到链表尾部 queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q); // timed表示是否有超时时间限制的阻塞 else if (timed) { nanos = deadline - System.nanoTime(); // 如果是有超时时间限制，但是超时时间是0，明显不合理，直接返回 if (nanos <= 0L) { removeWaiter(q); return state; } // 注意：这里就是调用LockSupport.parkNanos方法实现超时间的阻塞 LockSupport.parkNanos(this, nanos); } else // 注意：这里就是直接调用LockSupport的part方法进行阻塞 // 直到被调用LockSupport.unpark方法才能唤醒 LockSupport.park(this); } }
上面的awaitNode执行过程：
（1）首先判断一下当前任务的状态，如果任务状态大于COMPLETING，说明任务已经完成、或者已被取消、或者已被中断等等，此种状态不需要阻塞调用线程了
（2）如果当前任务状态是COMPLETING，说明任务运行完毕了，状态完成中，只差最后一步将state状态修改为NORMAL，就可以完成。
此种情况也不需要阻塞调用线程，让调用线程Thread.yield()让出cpu一段时间，毕竟任务很快就会完成了
（3）当任务状态为NEW，说明任务刚状态或者在运行中，此种情况不知道任务还要多久完成，所以调用线程只能阻塞等待了
（4）阻塞等待需要记录一下哪个线程被阻塞了，这个时候就需要一个等待队列了，就是存放被阻塞的线程，因为后续任务完成需要把他们一个个找出来唤醒
（5）阻塞等待有两种方式，有超时时间限制的底层就是调用LockSupport.parkNanos方法；没有超时时间限制的就是调用LockSupport.park方法
4.3.2 report 方法
private V report(int s) throws ExecutionException { // 这里就是将任务的运行结果outcome保存在变量v中 Object x = outcome; // 如果当前任务状态是NORMAl（已完成）则返回结果 if (s == NORMAL) return (V)x; // 如果当前状态大于等于CANCELLED表示被取消或者中断等异常状态，抛出异常 if (s >= CANCELLED) throw new CancellationException(); throw new ExecutionException((Throwable)x); }
我们画个图来整体理解下 get 方法：
上面的FutureTask的get方法已经讲解完毕了，get(long timeout, TimeUnit) 这个方法也是一样的，只是有超时时间限制而已。
4.4 run 方法
接下来我们就讲解FutureTask运行任务的run方法，FutureTask实现了RunnableFuture接口，而RunnableFuture接口继承了Runnable接口，所以FutureTask本身也是一个可以运行的任务。
public void run() { // 如果不是NEW状态，说明任务可能被取消了，拒绝执行 if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { // 获取Callable任务 Callable<V> c = callable; // 重新判断一下状态是否是NEW if (c != null && state == NEW) { // 任务的结果保存进入result变量中 V result; boolean ran; try { // 这里就是调用callable任务的call方法了，同时把结果保存到result变量中 result = c.call(); // 运行任务成功 ran = true; } catch (Throwable ex) { result = null; ran = false; setException(ex); } if (ran) // 这里的set其实就是将result保存到outcome变量中，方便以后对外输出结果 // 同时将任务的状态设置成为NORMAL（已完成） // 同时还会唤醒之前调用get方法而沉睡的线程， // 这些线程在等待队列waitnode中，需要一个个找出来调用LockSupport.unpark方法唤醒 set(result); } } finally { runner = null; // 如果被中断了，进行中断异常处理 int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }
4.4.1 set 方法
来继续看下set方法源码：
protected void set(V v) { // 首先运行完任务之后，将state变量通过cas设置为COMPLETING（完成中） if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { // 然后将任务的结果保存到outcome变量中 outcome = v; // 然后cas设置任务的状态为NORMAL（已完成） UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state // 这里就是一个个唤醒等待队列中的线程 finishCompletion(); } }
4.4.2 finishCompletion 方法
private void finishCompletion() { // 这里就是循环遍历waitNode等待队列 // 然后调用LockSupport.unpark方法一个个唤醒，没啥特别的 for (WaitNode q; (q = waiters) != null;) { if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) { for (;;) { Thread t = q.thread; if (t != null) { q.thread = null; LockSupport.unpark(t); } WaitNode next = q.next; if (next == null) break; q.next = null; // unlink to help gc q = next; } break; } } done(); callable = null; // to reduce footprint }
整个run方法，当FutureTask任务交给线程池运行的时候：
（1）执行FutureTask里面的run方法，会找到FutureTask里面的Callable任务执行
（2）将Callable任务执行完成后，返回结果保存进入result变量中
（3）任务执行完成，result结果保存到outcome变量中，设置state变量为NORMAL状态
我们看下FutureTask的get方法和run方法整合起来，画个图理解一下：
5 其它方法
我们最后看一下FutureTask的其它方法：
5.1 取消任务的方法cancel
public boolean cancel(boolean mayInterruptIfRunning) { // 如果当前状态是NEW，说明还未运行完成（运行完成是NORMAL） // 所以尝试将状态改为INTERRUPTING或者CANCELLED if (!(state == NEW && UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) return false; try { // in case call to interrupt throws exception if (mayInterruptIfRunning) { try { Thread t = runner; // 如果任务正在运行中，调用interrupt方法中断运行该任务的线程 if (t != null) t.interrupt(); } finally { // final state UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); } } } finally { // 唤醒一下等待队列的线程 finishCompletion(); } return true; }
由于运行run方法的时候会判断当前任务的状态是NEW的时候才会运行，所以这里如果将任务的状态修改了，变成非NEW，那么就不会继续执行run方法里面的逻辑了，所以任务就取消了。
6 小结
这节我们主要看了ThreadPoolExecutor的submit提交流程，就是构建一个FutureTask出来返回给调用者线程，然后线程池内部是调用execute方法去执行这个任务。同时也深入的分析了FutureTask内部的原理，如何执行任务，以及执行完成之后保存结果到outcome变量中，调用get方法的时候如果未完成会被阻塞，完成了就将outcome结果返回。有理解不对的地方欢迎指正哈。