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

32次阅读

共计 6731 个字符,预计需要花费 17 分钟才能阅读完成。

前言

上一篇内容写了 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);
        }
    }
}

首先是判断 FutureTaskstate状态,必须是 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);
    }
}

如果 FutureTaskstateNORMAL 或者 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 常见配置代码分析、最佳实践等手册,方便工作中使用,也是对之前看过的源码一种总结。敬请期待!
欢迎关注:

正文完
 0