线程池续你必须要知道的线程池submit实现原理之FutureTask

前言

上一篇内容写了 Java 中线程池的实现原理及源码分析，说好的是实实在在的大满足，想通过一篇文章让大家对线程池有个透彻的了解，但是文章写完总觉得还缺点什么?

上篇文章只提到线程提交的 execute() 方法，并没有讲解线程提交的 submit() 方法，submit()有一个返回值，可以获取线程执行的结果 Future<T>，这一讲就那深入学习下submit() 和FutureTask实现原理。

使用场景 & 示例

使用场景

我能想到的使用场景就是在并行计算的时候，例如一个方法中调用methodA()、methodB()，我们可以通过线程池异步去提交方法 A、B，然后在主线程中获取组装方法 A、B 计算后的结果，能够大大提升方法的吞吐量。

使用示例

/**
 * @author wangmeng
 * @date 2020/5/28 15:30
 */
public class FutureTaskTest {public static void main(String[] args) throws InterruptedException, ExecutionException {ExecutorService threadPool = Executors.newCachedThreadPool();

        System.out.println("==== 执行 FutureTask 线程任务 ====");
        Future<String> futureTask = threadPool.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {System.out.println("FutureTask 执行业务逻辑");
                Thread.sleep(2000);
                System.out.println("FutureTask 业务逻辑执行完毕！");
                return "欢迎关注: 一枝花算不算浪漫！";
            }
        });

        System.out.println("==== 执行主线程任务 ====");
        Thread.sleep(1000);
        boolean flag = true;
        while(flag){if(futureTask.isDone() && !futureTask.isCancelled()){System.out.println("FutureTask 异步任务执行结果：" + futureTask.get());
                flag = false;
            }
        }

        threadPool.shutdown();}
}

上面的使用很简单，submit()内部传递的实际上是个 Callable 接口，我们自己实现其中的 call() 方法，我们通过 futureTask 既可以获取到具体的返回值。

submit()实现原理

submit() 是也是提交任务到线程池，只是它可以获取任务返回结果，返回结果是通过 FutureTask 来实现的，先看下 ThreadPoolExecutor 中代码实现：

public class ThreadPoolExecutor extends AbstractExecutorService {public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
}

public abstract class AbstractExecutorService implements ExecutorService {protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable);
    }
}

提交任务还是执行 execute() 方法，只是 task 被包装成了 FutureTask，也就是在excute() 中启动线程后会执行 FutureTask.run() 方法。

再来具体看下它执行的完整链路图：

上图可以看到，执行任务并返回执行结果的核心逻辑实在 FutureTask 中，我们以 FutureTask.run/get 两个方法为突破口，一点点剖析FutureTask 的实现原理。

FutureTask 源码初探

先看下 FutureTask 中部分属性：

public class FutureTask<V> implements RunnableFuture<V> {
    private volatile int state;
    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;

    private Callable<V> callable;
    private Object outcome;
    private volatile Thread runner;
    private volatile WaitNode waiters;
}

1. state

当前 task 状态，共有 7 中类型。
NEW： 当前任务尚未执行
COMPLETING: 当前任务正在结束，尚未完全结束，一种临界状态
NORMAL：当前任务正常结束
EXCEPTIONAL: 当前任务执行过程中发生了异常。
CANCELLED: 当前任务被取消
INTERRUPTING: 当前任务中断中..
INTERRUPTED: 当前任务已中断

1. callble

用户提交任务传递的 Callable，自定义 call 方法，实现业务逻辑

1. outcome

任务结束时，outcome 保存执行结果或者异常信息。

1. runner

当前任务被线程执行期间，保存当前任务的线程对象引用

1. waiters

因为会有很多线程去 get 当前任务的结果，所以这里使用了一种 stack 数据结构来保存

FutureTask.run()实现原理

我们已经知道在线程池 runWorker() 中最终会调用到 FutureTask.run() 方法中，我们就来看下它的执行原理吧：

具体代码如下：

public class FutureTask<V> implements RunnableFuture<V> {public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    set(result);
            }
        } finally {
            runner = null;
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }
}

首先是判断 FutureTask 中state状态，必须是 NEW 才可以继续执行。

然后通过 CAS 修改 runner 引用为当前线程。

接着执行用户自定义的 call() 方法，将返回结果设置到 result 中，result 可能为正常返回也可能为异常信息。这里主要是调用set()/setException()

FutureTask.set()实现原理

set()方法的实现很简单，直接看下代码：

public class FutureTask<V> implements RunnableFuture<V> {protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
            finishCompletion();}
    }
}

将 call() 返回的数据赋值给全局变量 outcome 上，然后修改 state 状态为 NORMAL，最后调用finishCompletion() 来做挂起线程的唤醒操作，这个方法等到 get() 后面再来讲解。

FutureTask.get()实现原理

接着看下代码：

public class FutureTask<V> implements RunnableFuture<V> {public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }
}

如果 FutureTask 中state为 NORMAL 或者 COMPLETING，说明当前任务并没有执行完成，调用get() 方法会被阻塞，具体的阻塞逻辑在 awaitDone() 方法：

private int awaitDone(boolean timed, long nanos) throws InterruptedException {final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) {if (Thread.interrupted()) {removeWaiter(q);
                throw new InterruptedException();}

            int s = state;
            if (s > COMPLETING) {if (q != null)
                    q.thread = null;
                return s;
            }
            else if (s == COMPLETING)
                Thread.yield();
            else if (q == null)
                q = new WaitNode();
            else if (!queued)
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q);
            else if (timed) {nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {removeWaiter(q);
                    return state;
                }
                LockSupport.parkNanos(this, nanos);
            }
            else
                LockSupport.park(this);
        }
    }

这个方法可以说是 FutureTask 中最核心的方法了，一步步来分析：

如果 timed 不为空，这说明指定 nanos 时间还未返回结果，线程就会退出。

q是一个 WaitNode 对象，是将当前引用线程封装在一个 stack 数据结构中，WaitNode对象属性如下：

 static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
}

接着判断当前线程是否中断，如果中断则抛出中断异常。

下面就进入一轮轮的 if... else if... 判断逻辑，我们还是采用分支的方式去分析。

分支一：if (s > COMPLETING) {

此时 get() 方法已经有结果了，无论是正常返回的结果，还是异常、中断、取消等，此时直接返回 state 状态，然后执行 report() 方法。

分支二：else if (s == COMPLETING)

条件成立，说明当前任务接近完成状态，这里让当前线程再释放cpu，进行下一轮抢占cpu。

分支三：else if (q == null)

第一次自旋执行，WaitNode还没有初始化，初始化q=new WaitNode();

分支四：else if (!queued){

queued代表当前线程是否入栈，如果没有入栈则进行入栈操作，顺便将全局变量 waiters 指向栈顶元素。

分支五 / 六：LockSupport.park

如果设置了超时时间，则使用 parkNanos 来挂起当前线程，否则使用park()

经过这么一轮自旋循环后，如果执行 call() 还没有返回结果，那么调用 get() 方法的线程都会被挂起。

被挂起的线程会等待 run() 返回结果后依次唤醒，具体的执行逻辑在 finishCompletion() 中。

最终 stack 结构中数据如下：

FutureTask.finishCompletion()实现原理

具体实现代码如下：

private void finishCompletion() {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;
                q = next;
            }
            break;
        }
    }

    done();

    callable = null;
}

代码实现很简单，看过 get() 方法后，我们知道所有调用 get() 方法的线程，在 run() 还没有返回结果前，都会保存到一个有 WaitNode 构成的 statck 数据结构中，而且每个线程都会被挂起。

这里是遍历 waiters 栈顶元素，然后依次查询起 next 节点进行唤醒，唤醒后的节点接着会往后调用 report() 方法。

FutureTask.report()实现原理

具体代码如下：

private V report(int s) throws ExecutionException {
    Object x = outcome;
    if (s == NORMAL)
        return (V)x;
    if (s >= CANCELLED)
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

这个方法很简单，因为执行到了这里，说明当前 state 状态肯定大于 COMPLETING，判断如果是正常返回，那么返回outcome 数据。

如果 state 是取消状态，抛出 CancellationException 异常。

如果状态都不满足，则说明执行中出现了差错，直接抛出ExecutionException

FutureTask.cancel()实现原理

public boolean cancel(boolean mayInterruptIfRunning) {if (!(state == NEW && UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();} finally {UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {finishCompletion();
    }
    return true;
}

cancel()方法的逻辑很简单，就是修改 state 状态为 CANCELLED，然后调用finishCompletion() 来唤醒等待的线程。

这里如果mayInterruptIfRunning，就会先中断当前线程，然后再去唤醒等待的线程。

总结

FutureTask的实现原理其实很简单，每个方法基本上都画了一个简单的流程图来方便立即。

后面还打算分享一个 BlockingQueue 相关的源码解读，这样线程池也可以算是完结了。

在这之前可能会先分享一个 SpringCloud 常见配置代码分析、最佳实践等手册，方便工作中使用，也是对之前看过的源码一种总结。敬请期待！
欢迎关注：