FutureTask实现原理

╰半橙微兮° 2022-05-28 22:21 200阅读 0赞

# 执行过程 #

提交的Runnable任务在内部统一被转换为Callable任务。查看submit方法的返回值，为一个Future，实际上这个Futrue为FutureTask实例，通过此实例，调用get方法，可以阻塞当前线程，直到任务运行完毕，返回结果。

整个调用链条如下所示：

worker thread -> futureTask.run() -> callable.call() -> task.run()

如果提交的是Callable任务，则只有前面三个调用。

# FutureTask run 执行 #

![这里写图片描述][SouthEast]

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 {
                        // 调用callable call 方法
                        result = c.call();
                        ran = true;
                    } catch (Throwable ex) {
                        // 异常处理
                        result = null;
                        ran = false;
                        setException(ex);
                    }
                    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);
            }
        }

## 异常处理&结果处理 ##

/** * Causes this future to report an {@link ExecutionException} * with the given throwable as its cause, unless this future has * already been set or has been cancelled. * * <p>This method is invoked internally by the {@link #run} method * upon failure of the computation. * * @param t the cause of failure */
        protected void setException(Throwable t) {
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                // 执行结果
                outcome = t;
                // 设置Task 执行状态
                UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
                // 清理waiters栈
                finishCompletion();
            }
        }
    
        /** * Sets the result of this future to the given value unless * this future has already been set or has been cancelled. * * <p>This method is invoked internally by the {@link #run} method * upon successful completion of the computation. * * @param v the value */
        protected void set(V v) {
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                // 执行结果
                outcome = v;
                // 设置Task 执行状态
                UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
                // 清理waiters栈
                finishCompletion();
            }
        }

## FutureTask get 阻塞获取结果 ##

![这里写图片描述][SouthEast 1]

## 多线程获取结果 ##

执行 callable.call() 过程中**有多个线程调用了 同个FutureTask实例的get方法**，这时候，这些线程会被阻塞，存于一个栈中。

如上图所示：线程1，2，3调用FutureTask.get方法，由于**任务未执行结束，这时候，三个线程都将被阻塞休眠**，FutureTask中有一个栈，用于存放等待线程，栈顶指针为 FutureTask.waiters引用。

当任务执行完毕后，会调用finishCompletion方法，迭代唤醒整个栈中的线程，这时候，各个线程都将被唤醒，并且可以顺利拿到任务的执行结果(执行结果存于 FutureTask.outcome)。

## 代码分析 ##

/** * 设置超时时间等待FutureTask执行结果 * @throws CancellationException {@inheritDoc} */
        public V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            if (unit == null)
                throw new NullPointerException();
            int s = state;
            if (s <= COMPLETING &&
                // 加入等待栈并阻塞
                (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
                throw new TimeoutException();
            return report(s);
        }
    
    
        /** * Awaits completion or aborts on interrupt or timeout. * * @param timed true if use timed waits * @param nanos time to wait, if timed * @return state upon completion */
        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)
                    // 等待Node
                    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);
            }
        }

[SouthEast]: /images/20220529/6d0582312e144774a84a3a59d48fda59.png
[SouthEast 1]: /images/20220529/3056d3d9cb134648a188f45d5b6fb54d.png