JDK1.8源码分析：Future和FutureTask-任务异步执行结果

清疚 2022-03-19 09:16 180阅读 0赞

### 概述 ###

*  任务在Executor线程执行器当中是异步执行的，而有些任务是需要返回执行结果的，故在Executor派生接口ExecutorService接口中定义了带返回结果的提交方法submit，其中返回结果为Future接口的对象。
 *  Future接口主要提供了异步返回任务执行结果，取消任务执行，获取任务执行状态的功能，接口定义如下：
    
        public interface Future<V> { 
            // 取消任务执行
            // mayInterruptIfRunning用于控制如果任务正在执行，是否中断对应的执行线程来取消该任务
            // 成功cancel，则isCancelled和isDoned都返回true。
            boolean cancel(boolean mayInterruptIfRunning);
        
            // 任务是否取消
            boolean isCancelled();
        
            // 正常执行，被取消，异常退出都返回true
            boolean isDone();
        
            // 阻塞等待执行结果
            // CancellationException：任务被取消
            // ExecutionException：任务执行异常
            // InterruptedException：该等待结果线程被中断
            V get() throws InterruptedException, ExecutionException;
        
            // 阻塞等待执行结果指定时间，除了以上异常，
            // TimeoutException：等待超时
            V get(long timeout, TimeUnit unit)
                throws InterruptedException, ExecutionException, TimeoutException;
        }

### FutureTask ###

*  Future接口的主要实现类为FutureTask，FutureTask同时实现了Runnable和Future接口，故对应的对象实例可以作为任务提交到Executor线程执行器中执行，然后通过自身来获取任务执行结果或者取消任务执行：
 *  即FutureTask的对象实例被Executor线程执行器内部线程池的某个工作线程和调用get方法等待获取结果的应用主线程所共享，故Executor内部线程池的工作线程在执行完这个任务后，可以通知和唤醒调用get阻塞等待执行结果的应用主线程，应用主线程也可以取消该任务的执行，然后通知工作线程。
    
        // 被线程池ExecutorService和调用get方法等待节点的主线程共享
        // 线程池的执行线程worker调用run方法；等待结果的线程调用get等待结果，
        // get通过自旋和LockSupport.park来先休眠，后等待执行线程在执行完run方法，调用LockSupport.unpark来唤醒，
        // 调用get的线程被唤醒后，由于存在自旋，故再次检查是否有结果了，由结果则可以返回了。
        public class FutureTask<V> implements RunnableFuture<V> { 
            ...
        }
        
        public interface RunnableFuture<V> extends Runnable, Future<V> { 
            /** * Sets this Future to the result of its computation * unless it has been cancelled. */
            void run();
        }
 *  在FutureTask中定义了volatile修饰的状态变量state来进行Executor中的工作线程和应用主线程之间的交互，即工作线程产生任务执行结果，通知应用主线程获取；应用主线程请求取消任务执行，通知工作线程停止该任务执行。在内部实现中通过将state与以下状态常量进行大小比较来获取任务执行情况，如是正常执行成功还是异常退出，被取消等。
    
        /* * 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;
        
        /** The result to return or exception to throw from get() */
        private Object outcome; // non-volatile, protected by state reads/writes
        /** The thread running the callable; CASed during run() */
        private volatile Thread runner;

#### 提交任务到Executor线程执行器 ####

*  提交任务到Executor线程执行器，对应AbstractExecutorService的submit方法实现如下：在submit中创建了一个FutureTask对象来包装应用定义的Runnable接口实现类task，调用execute将该对象交给Executor线程执行器执行，然后返回该对象引用给应用主线程。
    
        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { 
            return new FutureTask<T>(callable);
        }
        
        /** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */
        public Future<?> submit(Runnable task) { 
            if (task == null) throw new NullPointerException();
            // 将应用代码中的task包装成一个FutureTask，然后交给线程池执行
            // 线程池调度一个线程来执行FutureTask的run，在FutureTask的run中调用task的run
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            execute(ftask);
            // 返回这个包装了task的FutureTask，在应用代码中可以通过get来获取执行结果
            return ftask;
        }

#### 应用主线程调用get等待执行结果 ####

*  在FutureTask中的get方法实现如下：
    
        public V get() throws InterruptedException, ExecutionException { 
            int s = state;
            if (s <= COMPLETING)
                // 休眠等待执行结果
                s = awaitDone(false, 0L);
            // 休眠返回了，调用report获取结果
            return report(s);
        }
 *  任务状态state小于等于COMPLETING表示任务还没开始执行，则应用主线程调用awaitDone阻塞休眠，等待Executor的工作线程执行任务并通知唤醒该应用主线程。具体过程如下：
    
        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;
                // 在set中，将状态从COMPLETING变成了NORMAL，并调用finishCompletion唤醒等待线程集合的所有线程，
                // NORMAL大于COMPLETING，故可以返回了。
                // 另外其他不是正常执行而结束的情况，如CANCELED, INTERRUPTED, EXCEPTIONAL等也是在这里返回
                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)
                    // 下一轮自旋到下一个else if (!queued)，然后入队waiters
                    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
                    // 被finishCompletion唤醒，自旋回到for循环重新往下执行
                    LockSupport.park(this);
            }
        }
 *  在FutureTask内部维护了一个单向链表waiters，用于存放当前等待该任务执行结果的线程，在任务执行完成时，遍历该链表，唤醒每个等待线程。
    
        /** Treiber stack of waiting threads */
        private volatile WaitNode waiters;
        static final class WaitNode { 
            volatile Thread thread;
            volatile WaitNode next;
            // thread为调用get方法的线程，通常为主线程
            WaitNode() {  thread = Thread.currentThread(); }
        }

#### Executor工作线程执行任务 ####

*  Executor的工作线程执行该任务时，会调用该任务的run方法，即FutureTask的run方法，如下为FutureTask的run方法定义：首先检查任务状态state是否为NEW，是，即还没执行过也没有被取消等，则进行往下执行。执行完成之后，产生执行结果result，调用set方法来处理这个结果。
    
        public void run() { 
            // 赋值runner，指向线程池中当前执行这个task的线程
            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 { 
                        // 应用代码的task的run方法执行
                        result = c.call();
                        ran = true;
                    } catch (Throwable ex) { 
                        result = null;
                        ran = false;
                        setException(ex);
                    }
                    // 执行成功设置结果result，并通知get方法
                    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);
            }
        }
 *  set方法的定义如下：将执行结果赋值给FutureTask的成员变量outcome，更新任务执行状态state为NORMAL，最后调用finishCompletion通知所有等待这个任务执行结果的线程。
    
        protected void set(V v) { 
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { 
                outcome = v;
                // 修改状态为NORMAL，NORMAL大于COMPLETING
                UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
                // 唤醒调用get方法等待结果的线程
                finishCompletion();
            }
        }
 *  finishCompletion的实现如下：遍历任务等待线程链表，使用LockSupport.unpart唤醒对应的线程，然后将该等待线程从链表中移除。
    
        private void finishCompletion() { 
            // assert state > COMPLETING;
            // waiters为调用了get方法的线程集合，通常为主线程
            for (WaitNode q; (q = waiters) != null;) { 
                if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) { 
                    for (;;) { 
                        Thread t = q.thread;
                        if (t != null) { 
                            q.thread = null;
                            // 唤醒等待线程，即调用了get方法的线程
                            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
        }
 *  然后回到get方法，应用主线程从awaitDone阻塞返回后，通过report方法来检测执行状态并返回任务执行结果。
    
        public V get() throws InterruptedException, ExecutionException { 
            int s = state;
            if (s <= COMPLETING)
                // 休眠等待执行结果
                s = awaitDone(false, 0L);
            // 休眠返回了，调用report获取结果
            return report(s);
        }
        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);
        }

#### 应用主线程取消任务 ####

*  在应用主线程中，可以通过调用FutureTask的cancel方法来取消该任务的执行，cancel方法的定义如下：主要是更新任务的状态state为INTERRUPTING或者CANCELLED，然后根据mayInterruptIfRunning来控制如果该任务已经在执行，是否中断对应的工作线程来中止该任务的执行，最后调用finishCompletion方法来唤醒等待这个任务执行结果的线程，避免该任务被取消后，这些线程还在阻塞等待结果。
    
        public boolean cancel(boolean mayInterruptIfRunning) { 
            // 只能取消还没被执行的task，即state为NEW表示还没执行
            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;
                        // 中断执行线程，这样线程池在之后会补回来一个执行线程
                        if (t != null)
                            t.interrupt();
                    } finally {  // final state
                        UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                    }
                }
            } finally { 
                // 唤醒调用了get方法等待获取该任务的结果的线程，避免该任务被取消了，这些线程还在等待。
                finishCompletion();
            }
            return true;
        }
 *  在Executor的工作线程执行这个任务时，会调用FutureTask的run方法，在run方法中是先检查任务的状态state，如果发现不是NEW，即可能是CANCELLED，INTERRUPTING等，则直接返回，退出该任务的执行。
    
        public void run() { 
            // 赋值runner，指向线程池中当前执行这个task的线程
            if (state != NEW ||
                !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                             null, Thread.currentThread()))
                return;
            }
            
            ...
            
        }