FutureTask源码分析

末蓝、 2021-09-21 22:16 409阅读 0赞

1.  可以使用FutureTask来创建一个线程，用来异步执行任务，并且可以保证并发环境下只执行一次(run方法中，通过CAS设置状态，runner指向当前线程来保证)，并且可以获取任务执行结果(获取结果get是阻塞的，知道任务完成返回结果)。FutureTask实现了Runnable接口，并重写了run方法。在FutureTask中包含一个Callable接口对象，创建FutureTask实例时，要重写Callable的call方法，call方法就是线程执行的具体逻辑。Callable类似于Runnable，区别在于Callable具有返回值(Runnable接口的run方法无法抛出check exception)。
    
    1.  实际的执行过程，线程的执行实际都是通过Runnable接口的run方法来实现的，当线程启动执行时，会执行run方法。FutureTask实现了Runnable，那么FutureTask也具有Runnable的特性，可以用来执行线程。并且FutureTask实现了Future接口(配合FutureTask的Callable属性，获取线程执行情况，并且可以获取返回值)，使其具有返回结果值得功能。当线程启动执行时，先执行run方法，在run方法中，会调用Callable的call方法，即具体线程执行逻辑，这个call方法是在创建FutureTask时指定的，重写了FutureTask的Callable属性的call方法，而在run方法中就是调用的call方法来执行的，当执行完成call方法逻辑时，通过Future的set方法(FutureTask重写)，来保存线程执行结果，用于使用get方法来获取返回值。
    2.  FutureTask是对Runnable的增强，他实现了RunnableFuture(实现了Future和Runnable)。所以他是一个Runnable，也就是说他可以用来代替Runnable使用。因为Runnable没有返回值，而且不能抛出check exception必须try catch处理。对于返回值FutureTask使用Callable和Future接口实现，可以获取线程的执行结果，取消线程，也可以检测线程的执行情况，例如是否取消，是否已完成。而且Callable也解决了不能抛出check exception的问题。
2.  继承关系
    
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JpZ2x4bA_size_16_color_FFFFFF_t_70][]
    
    通过实现Runnable接口，可以通过Thread，线程池来执行。  
    通过实现Future接口，可以获取任务的执行结果。
3.  FutureTask结构图  
     
    
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JpZ2x4bA_size_16_color_FFFFFF_t_70 1][]
    
     
4.  FutureTask状态
    
    1.  FutureTask定义一个volatile属性state，表示FutureTask生命周期。
        
        1.  NEW 初始状态。
        2.  COMPLETING 任务已经执行完(正常执行完成或抛出异常)，准备赋值结果。
        3.  NORMAL 任务已经正常执行完成，并且已经将任务返回值赋值到结果Future中。
        4.  EXCEPTIONAL 任务执行失败，并将异常赋值到结果Future中。
        5.  CANCELLED 取消
        6.  INTERRUPTING 准备尝试中断任务的线程。
        7.  INTERRUPTED 对执行任务的线程进行中断(未必中断成功)
    2.  ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JpZ2x4bA_size_16_color_FFFFFF_t_70 2][]
    3.      /**
                 * The run state of this task, initially NEW.  The run state
                 * transitions to a terminal state only in methods set,
                 * setException, and cancel.  During completion, state may take on
                 * transient values of COMPLETING (while outcome is being set) or
                 * INTERRUPTING (only while interrupting the runner to satisfy a
                 * cancel(true)). Transitions from these intermediate to final
                 * states use cheaper ordered/lazy writes because values are unique
                 * and cannot be further modified.
                 *
                 * 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;
            	// 最终状态,表示任务正常执行完成，并且已经设置任务执行结果,将任务执行返回值存储在outcome对象中,作为任务执行结果。
                private static final int NORMAL       = 2;
            	// 最终状态,表示任务执行中抛出异常，并且已经设置任务执行结果,将任务抛出的异常存储在outcome对象中,作为任务执行结果。
                private static final int EXCEPTIONAL  = 3;
            	// 最终状态,表示任务被直接取消
                private static final int CANCELLED    = 4;
            	// 中间状态,表示任务正在尝试被取消
                private static final int INTERRUPTING = 5;
            	// 最终状态,表示任务已经被取消
                private static final int INTERRUPTED  = 6;
        
         
    4.  1.  FutureTask的几个方法对应了FutureTask的状态的变化。
            
            1.  通过构造方法FutureTask(Callable<T> callable)或者FutureTask(Runnable runnable, V result)创建任务，此时状态为NEW。
            2.  任务被线程调度执行，进入线程的run方法，执行任务。任务执行完成(调用Callable的call方法，执行完call方法体，或者再call方法中抛出异常)，此时为COMPLETING状态(中建状态)。如果正常结束，调用set方法设置任务执行结果，此时为NORMAL状态；如果抛出异常，调用setException方法设置任务执行结果，此时为EXCEPTIONAL状态。NORMAL和EXCEPTIONAL为最终状态。、
            3.  在任务调度过程中，如果将任务取消，调用cancel方法，则进入INTERRUPTING状态。如果cancel(true)，进入INTERRUPTED状态；如果cancel(false)，进入CANCELLED状态。
5.  run方法
    
        /**
         * run方法，线程执行体。
         * 在通过FutureTask创建线程体，实现的Callable接口call方法，call方法记录的具体任务执行逻辑。
         * 启动线程后，首先调用FutureTask重写的run方法，线程的执行入口，在run方法中调用真正的任务逻辑
         * Callable的call方法。
        */
        public void run() {
            	// 首先判断，任务状态，如果不是NEW状态，表示任务已经运行，直接返回。
                if (state != NEW ||
                    // 其次，设置任务的执行线程为当前线程
                    // this表示当前任务，runnerOffset表示在内存中runner属性的偏移量
                    // (同过this地址加上runnerOffset偏移量，确定指向runner属性)
                    // 通过UNSAFE尝试改变runner属性为当前线程,期待当前runner对象的值为null。
                    // 如果修改失败(返回false)，直接返回。
                    // UNSAFE 通过反射FutureTask类，获取runner字段Field对象，通过UNSAFE.objectFieldOffset方法
                    // 获取到runner字段在FutureTask类中的偏移量。
                    // UNSAFEcompareAndSwap等方法通过当前对象地址+偏移量 指向被修改的字段，再传入期望值和修改值，
                    // 尝试修改，修改成功返回true，否则返回false。
                    !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                                 null, Thread.currentThread()))
                    // 在多线中并发抢占任务时，只允许一个线程执行该任务。其他的线程会进入等待队列中。
                    // 使用WaitNode waiters存储, WaitNode是FutureTask的内部类, 链表结构。
                    // 在任务完成时需要释放所有等待的线程, finishCompletion方法。
                    return;
                try {
                    // 引用成员变量 具体任务逻辑callable对象
                    Callable<V> c = callable;
                    // 双重检查，判断任务逻辑对象callable是否为空，并且任务当前状态是否是NEW状态
                    if (c != null && state == NEW) {
                        // 记录任务结果, 即Callable任务逻辑的返回值。
                        V result;
                        // 记录任务是否正常结束, 如果Callable的call方法这正常结束返回,则为true;
                        // 若call方法抛出异常, 则为false。
                        boolean ran;
                        try {
                            // 执行实际任务逻辑, Callable的call方法, 即创建FutureTask传入的任务逻辑。
                            result = c.call();
                            // 任务逻辑正常结束返回, 返回true。
                            ran = true;
                        } catch (Throwable ex) {
                            // 任务逻辑抛出异常
                            // 异常结束, 执行结果为null
                            result = null;
                            // 任务逻辑正常结束返回, 返回false。
                            ran = false;
                            // 设置任务执行结果, 并设置任务最终状态为EXCEPTIONAL
                            setException(ex);
                        }
                        if (ran)
                            // 任务正常执行返回。设置任务执行结果, 并设置任务最终状态为NORMAL
                            set(result);
                    }
                } finally {
                    // runner属性, 记录执行任务的线程, 执行任务过程中runner必须非空, 直到任务状态发生改变。
                    // runner must be non-null until state is settled to
                    // 防止并发调用run方法
                    // prevent concurrent calls to run()
                    
                   	// 多个线程同时调用任务，只会有一个任务获取到线程(之前的判断)。这里通过runner是否为空来控制多线程并发访问。
                    // runner==null时, 线程获取到任务, 执行Callable的call方法, 其他获取到任务的线程直接return, 不会再
                    // 执行run方法了。防止并发执行任务逻辑call方法。
                    runner = null;
                    // state must be re-read after nulling runner to prevent
                    // leaked interrupts
                    int s = state;
                    if (s >= INTERRUPTING)
                        handlePossibleCancellationInterrupt(s);
                }
            }
    
     
6.  set、setException方法  
     
    
        // 当任务正常结束时, 设置状态, 任务执行结果, 执行后续操作
        	protected void set(V v) {
                // 将状态从NEW状态改为中间状态COMPLETING状态, 表示任务已经运行完成, 正在处理后续结果。
                if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                    // 赋值任务执行结果
                    outcome = v;
                    // 将状态修改为最终状态NORMAL, 表示任务已经正常完成, 并且返回结果。
                    UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
                    // 执行后续操作。
                    finishCompletion();
                }
            }
        	// 当任务抛出异常时, 设置状态, 任务执行结果, 执行后续操作。
            protected void setException(Throwable t) {
                // 将状态从NEW状态改为中间状态COMPLETING状态, 表示任务已经运行完成, 正在处理后续结果。
                if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                    // 赋值任务异常结果
                    outcome = t;
                    // 将状态改为最终状态EXCEPTIONAL, 表示任务异常结束, 并且返回结果。
                    UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
                    // 执行后续操作。
                    finishCompletion();
                }
            }
    
      
     
7.  cancel方法  
     
    
        /**
         * 尝试取消执行此任务。如果任务已经完成, 已经被取消或由于某种其他原因而无法取消, 则此尝试将失败, 返回false。
         * 当cancel时此任务尚未启动，则此任务不应运行。 如果任务已经开始，那么mayInterruptIfRunning(true或者false)参数确定是否中断执行此任务的线程。
         * cancel(true)方法只是想执行任务的线程发送中断请求(设置线程的中断标志), 是否能够中断, 在哪一步中断, 取决于线程本身。
         * 实际使用可以配合Thread.currentThread().isInterrupted()来进行判断。
         * 任务在有线程执行前调用了cancel方法，那么任务状态会变成最终状态CANCELED(调用cancel(false)方法), 或者INTERRUPTED(调用cancel(true)方法)。
         * 任务未运行直接进入最终状态，之后即使通过启动线程运行该任务, 该任务也不会执行, 因为该任务不是NEW状态了。
         * cancel(true), 只是调用现Thread的interrupt方法, 设置线程的中断标志。并不能真正的中断执行任务的线程。最终结果, 要么成功中断线程, 要么线程运行完成,
         * 但最终状态都是CANCELED或者INTERRUPTED, 表示任务已经被取消。
         */
        	public boolean cancel(boolean mayInterruptIfRunning) {
                //如果状态不为NEW，且无法将状态更新为INTERRUPTING或CANCELLED，则直接返回取消失败
                // 通过UNSAFE类，尝试将NEW状态的FutureTask改为INTERRUPTING状态，表示FutureTask现在正在尝试cancel，
                // 正处于中间状态，直到成功取消任务，状态变为CANCELLED或者INTERRUPTED。
                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;
                            // 如果没有线程来执行该任务, runner==null, 那么直接返回
                            if (t != null)
                                // 通过调用线程的interrupt方法，设置线程中断标志。尝试中断阻塞的线程。
                                t.interrupt();
                        } finally { // final state
                            // 将任务状态改为INTERRUPTED
                            UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                        }
                    }
                } finally {
                    // 任务由中间状态到最终状态变更时，都需进行的操作。
                    // 调用回调方法 done, 任务完成时将任务结果插入结果队列BlockingQueue。
                    finishCompletion();
                }
                return true;
            }
    
      
     
8.  finishCompletion方法  
     
    
        // 任务完成的后续工作
        	// 唤醒所有等待线程节点, 释放任务等待链表, 释放内存资源。
        	// 调用回调方法done, 当任务完成时执行某些操作, done方法由子类实现。
        	private void finishCompletion() {
                // assert state > COMPLETING;
                // 此时, 任务状态一定大于COMPLETING(要么完成NORMAL, EXCEPTIONAL, 要么取消CANCELED, INTERRUTPED)
                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; // unlink to help gc
                            q = next;
                        }
                        break;
                    }
                }
        		// 调用回调方法, 又子类决定。FutureTask的done是空方法。
                done();
        		
                callable = null;        // to reduce footprint
            }
    
      
     
9.  get方法  
     
    
        /**
             * get方法, 获取任务的执行结果, 可以设置等待结果的超时时间。
             * 只有状态在NEW、COMPLETING、NORMAL才允许获取结果, 其他状态返回异常。
             */
        	public V get() throws InterruptedException, ExecutionException {
                // 获取当前状态
                int s = state;
                // 状态值小于COMPLETING时, 才获取状态。
                if (s <= COMPLETING)
                    // 调用awaitDone方法, 是否设置超时时间false, 超时时间0L。阻塞等待获取结果。返回值是任务状态。
                    s = awaitDone(false, 0L);
                // 返回结果
                return report(s);
            }
        
            public V get(long timeout, TimeUnit unit)
                throws InterruptedException, ExecutionException, TimeoutException {
                if (unit == null)
                    throw new NullPointerException();
                int s = state;
                if (s <= COMPLETING &&
                    // 调用awaitDone方法, 是否设置超时时间true, 超时时间timeout(转换为纳秒)。阻塞等待获取结果。
                    // 返回值是任务状态。
                    (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
                    throw new TimeoutException();
                // 返回结果
                return report(s);
            }
    
      
     
10. awaitDone方法  
     
    
        // 等待任务完成, 或者任务被取消中断, 或者超时
        // 返回状态
        // 执行get获取结果的线程, 与执行任务的线程, 不是同一个线程
        private int awaitDone(boolean timed, long nanos)
                throws InterruptedException {
            // 计算超时时间, 单位:纳秒
                final long deadline = timed ? System.nanoTime() + nanos : 0L;
            // 当前线程等待节点, 如果当前线程未获取到任务结果, 则会被加入任务的等待链表中, 进行等待。
                WaitNode q = null;
            // 是否已经加入链表
                boolean queued = false;
            // 死循环等待任务完成(正常执行完成, 中断, 超时)
                for (;;) {
                    // 判断当前执行任务的线程是否调用cancel(true)方法,进行中断任务。中断则返回InterruptedException异常。
                    if (Thread.interrupted()) {
                        // ?
                        removeWaiter(q);
                        throw new InterruptedException();
                    }
        			// 获取任务的状态
                    int s = state;
                    // 如果任务状态大于COMPLETING, 表示任务可能已经正常完成, 抛出异常, 被取消。返回状态值。
                    if (s > COMPLETING) {
                        //  ?
                        if (q != null)
                            q.thread = null;
                        return s;
                    }
                    else if (s == COMPLETING) // cannot time out yet
                        // 任务已经处于中间状态COMPLETING, 表示任务即将完成(正常完成, 抛出异常)
                        // 任务正在执行, 不计算超时时间
                        // 当前获取任务结果的线程, 重新进入就绪状态, 使执行任务的线程有机会获取CPU资源继续执行, 设置任务执行结果。
                        Thread.yield();
                    else if (q == null)
                        // 如果线程还没有完成
                        // 创建等待节点, 指向当前线程。
                        // 多线程并发获取结果时, 在任务没有执行完成时, 都会加入到当代链表中。
                        // 在下一次循环中, 如果任务还未执行完成, 将当前线程等待节点加入等待链表中。
                        q = new WaitNode();
                    else if (!queued)
                        // 如果还没有加入等待链表, 将当前线程节点加入到等待链表中。
                        // CAS方法返回是否成功修改, 即是否成功加入等待链表。
                        // 将当前线程等待节点加入到链表头部。
                        queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                             q.next = waiters, q);
                    // 判断获取结果是否超时
                    else if (timed) {
                        // 超时时间-系统当前时间
                        nanos = deadline - System.nanoTime();
                        // 结果<0, 表示已经超时, 返回当前任务状态。
                        if (nanos <= 0L) {
                            // 如果当前获取结果线程超时, 那么将当前线程等待节点从等待链表中移除。返回任务状态。
                            removeWaiter(q);
                            return state;
                        }
                        // 还未超时, 则阻塞等待剩余超时时间, 等待时间到, 自动唤醒。
                        LockSupport.parkNanos(this, nanos);
                    }
                    else
                        // 不计算超时时间
                        // 让当前线程进行阻塞等待, 直至任务完成(正常完成, 跑出异常, 任务被取消)时被唤醒。
        				// 唤醒后, 继续循环获取结果。然后要么返回正常执行结果, 要么跑出异常。
                        LockSupport.park(this);
                }
            }
    
      
     
11. report方法  
     
    
        /**
             * Returns result or throws exception for completed task.
             * (翻译源码注释)对于最终状态的任务, 返回结果或者抛出异常
             */
        	private V report(int s) throws ExecutionException {
                // 获取结果
                // 正常完成, 状态=NORMAL, outcome=任务执行结果(call方法返回值)
                // 抛出异常, 状态=EXCEPTIONAL, outcome=抛出的异常
                // 其他状态, outcome=null
                Object x = outcome;
                if (s == NORMAL)
                    // 正常完成, 直接返回结果
                    return (V)x;
                if (s >= CANCELLED)
                    // 如果任务调用了cancel方法, 正在尝试中断或者已经被中断。那么返回中断异常CancellationException, 
                    // 表示该任务已经被取消, 没有返回值
                    throw new CancellationException();
                // 任务异常结果, 抛出异常。
                throw new ExecutionException((Throwable)x);
            }
    
      
     
12. removeWaiter方法  
     
    
        // 从等待链表中移除某个获取结果的线程等待节点
        	private void removeWaiter(WaitNode node) {
                if (node != null) {
                    node.thread = null;
                    retry:
                    for (;;) {          // restart on removeWaiter race
                        for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                            s = q.next;
                            if (q.thread != null)
                                pred = q;
                            else if (pred != null) {
                                pred.next = s;
                                if (pred.thread == null) // check for race
                                    continue retry;
                            }
                            else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                                  q, s))
                                continue retry;
                        }
                        break;
                    }
                }
            }
    
      
     
13. 相关问题
    
    1.  cancel(true)和cancel(false)区别？
        
        1.  false表示，如果任务没有运行，就不要运行了，最终状态为CANCELED；  
            如果任务已经运行，但未运行完成，那么就让任务运行完，最终状态为CANCELED。
        2.  true表示，如果任务没有运行，就不要运行了，最终状态为INTERRUPTED；  
            如果任务已经运行，但未运行完成，那么就尝试中断线程(Thread.interrupt)，并不能一定中断线程。
    2.  什么时候使用cancel(true)
        
        1.  如果执行任务的线程是由Executor创建的，他实现了一种中断策略使得任务可以通过中断被取消，此时可以使用true中断线程。
    3.  FutureTask WaitNode 有什么用？
        
        1.  调用get方法获取任务结果的所有线程，由于run方法没有执行完成，导致获取结果的线程发生等待。waiters对象保存着这些线程，WaitNode结构使这些线程形成一个链表，头结点表示最后调用get方法的线程等待节点。
        2.  当任务执行完成(run方法执行完成)、任务执行过程中抛出异常、或者调用cancel方法取消任务。就需要让等待链表waiters退出等待
    4.  FutureTask使用不当可能会造成线程一直等待问题？
        
        1.  线程池使用FutureTask的时候，如果拒绝策略设置为DiscardPolicy(丢弃任务，但是不抛出异常)或DiscardOldestPolicy(丢弃队列最前面的任务，然后重新提交被拒绝的任务)，并且调用了被拒绝任务的无参get方法，那么调用线程会一直阻塞。
        2.  原因：当任务提交后被拒绝时，任务处于NEW状态，此时调用任务的get方法，调用线程会被awaitDone方法加入到等待链表中waiters，并且会一直阻塞住(LockSupport.park)，直到任务run方法结束，或者调用了cancel方法取消任务时，才会唤醒等待链表中的线程(LockSupport.unpark)，完成获取结果操作。但是通过拒绝策略的任务，并不会执行也不会取消，所以没有唤醒获取结果线程的机会，所以获取结果的线程会一直处于阻塞队列中，一直占用资源。简单的说就是任务被拒绝策略丢弃后，有现成获取该任务的结果，那么获取结果的线程就会一直等待。
        3.  解决办法：
            
            1.  使用带有超时时间的get方法获取任务结果。
            2.  使用自定拒绝策略。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JpZ2x4bA_size_16_color_FFFFFF_t_70]: /images/20210920/6d3829391f7148048ff1092bc61b8a92.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JpZ2x4bA_size_16_color_FFFFFF_t_70 1]: /images/20210920/ae77f68ae9a44be0a8ce64c8040f594a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JpZ2x4bA_size_16_color_FFFFFF_t_70 2]: /images/20210920/ead1e62f9f074083ad07a5b9bd4febda.png