FutureTask 原理

二十、FutureTask 原理

20.1 一个例子

    static class Task implements Callable<Integer> {
        @Override
        public Integer call() throws Exception {
            System.out.println("子线程在进行计算");
            Thread.sleep(1000);
            int sum = 0;
            for (int i = 0; i < 100; i++)
                sum += i;
            return sum;
        }
    }

    public static void main(String[] args) throws InterruptedException {

        ExecutorService executor = Executors.newCachedThreadPool();
        Task task = new Task();
        FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
        executor.submit(futureTask);

        System.out.println("主线程在执行任务");

        try {
            System.out.println("task运行结果" + futureTask.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        System.out.println("所有任务执行完毕");
        executor.shutdown();

    }

如上代码主线程会在futureTask.get()出阻塞直到task任务执行完毕,并且会返回结果。

20.1 原理

先看下类图结构

image.png

FutureTask 内部有一个state用来展示任务的状态,并且是volatile修饰的:

/** Possible state transitions:
 * NEW -> COMPLETING -> NORMAL 正常的状态转移
 * NEW -> COMPLETING -> EXCEPTIONAL 异常
 * NEW -> CANCELLED 取消
 * NEW -> INTERRUPTING -> INTERRUPTED 中断
 */
 
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;

其中构造FutureTask实例时候状态为new

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;      
    }

把FutureTask提交到线程池或者线程执行start时候会调用run方法:

public void run() {

    //如果当前不是new状态,或者当前cas设置当前线程失败则返回,只有一个线程可以成功。
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        //当前状态为new 则调用任务的call方法执行任务
        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);完成NEW -> COMPLETING -> EXCEPTIONAL 状态转移
            }

            //执行任务成功则保存结果更新状态,unpark所有等待线程。
            if (ran)
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}


protected void set(V v) {
    //状态从new->COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        //状态从COMPLETING-》NORMAL
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        //unpark所有等待线程。
        finishCompletion();
    }
}

任务提交后,会调用 get方法获取结果,这个get方法是阻塞的。

   public V get() throws InterruptedException, ExecutionException {
        int s = state;
        //如果当前状态是new或者COMPLETING则等待,因为位normal或者exceptional时候才说明数据计算完成了。
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }
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) // cannot time out yet
            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;
            }
            //否者设置park超时时间
            LockSupport.parkNanos(this, nanos);
        }
        else
            //直接挂起当前线程
            LockSupport.park(this);
    }
}
    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);
    }

在submit任务后还可以调用futuretask的cancel来取消任务:

    public boolean cancel(boolean mayInterruptIfRunning) {
        //只有任务是new的才能取消
        if (state != NEW)
            return false;
       //运行时允许中断
        if (mayInterruptIfRunning) {
           //完成new->INTERRUPTING
            if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
                return false;
            Thread t = runner;
            if (t != null)
                t.interrupt();
            //完成INTERRUPTING->INTERRUPTED
            UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
        }
       //不允许中断则直接new->CANCELLED
        else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
            return false;
        finishCompletion();
        return true;
    }

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Linyb极客之路

Java高级特性——注解,这也许是最简单易懂的文章了

博主在初学注解的时候看到网上的介绍大部分都是直接介绍用法或者功能,没有实际的应用场景,篇幅又很长导致学习的时候难以理解其意图,而且学完就忘QAQ。本篇文章中我将...

12420
来自专栏编码小白

tomcat请求处理分析(一) 启动container实例

1.1.1  启动container实例 其主要是进行了生命周期中一系列的操作之后调用StandardEngine中的 startInternal方法,不难看出...

38860
来自专栏Java架构师学习

Zookeeper-watcher机制源码分析(二)

其大致流程如下   ① 通过传入的path(节点路径)从watchTable获取相应的watcher集合,进入②

16110
来自专栏Android 开发学习

JsBridge 源码分析

19630
来自专栏会跳舞的机器人

java并发编程的艺术第十章——Executor框架

Executor框架的主要成员:ThreadPoolExecutor、ScheduledThreadPoolExecutor、Future接口、Runnable...

12520
来自专栏Android相关

Java线程池---ShutDown以及ShutDownNow解析

初始化一个有序的关闭,之前提交的任务都会被执行,但是新提交的任务则不会被允许放入任务队列中。如果之前被调用过了的话,那么再次调用也没什么用。这个方法不会等待之前...

12020
来自专栏Java编程技术

Java中调度线程池ScheduledThreadPoolExecutor原理探究

前面讲解过Java中线程池ThreadPoolExecutor原理探究,ThreadPoolExecutor是Executors中一部分功能,下面来介绍另外一部...

16120
来自专栏林冠宏的技术文章

java 线程池ThreadPoolExecutor 如何与 AsyncTask() 组合使用。

转载请声明出处谢谢!https://cloud.tencent.com/developer/user/1148436/activities 这里主要使用Exec...

22160
来自专栏Flutter入门

Weex是如何在Android客户端上跑起来的

Weex可以通过自己设计的DSL,书写.we文件或者.vue文件来开发界面,整个页面书写分成了3段,template、style、script,借鉴了成熟的MV...

52350
来自专栏猿天地

面试-线程池的成长之路

线程池是一种多线程处理形式,处理过程中将任务提交到线程池,任务的执行交由线程池来管理。

17420

扫码关注云+社区

领取腾讯云代金券