FutureTask源码解析(JDK1.8)

Posted 2022-03-22 liuhmmjj

在Java中一般通过继承Thread类或者实现Runnable接口这两种方式来创建多线程，但是这两种方式都有个缺陷，就是不能在执行完成后获取执行的结果，因此Java 1.5之后提供了Callable和Future接口，通过它们就可以在任务执行完毕之后得到任务的执行结果。

下面分析其源码：

Callable接口

//Runnable是没有返回结果的任务，而Callable则是有返回结果的任务
public interface Callable<V> 
    /**
     * 有返回结果，并且可能抛出异常
     */
    V call() throws Exception;

可以看到Callable是个泛型接口，泛型V就是要call()方法返回的类型。Callable接口和Runnable接口很像，都可以被另外一个线程执行，但是正如前面所说的，Runnable不会返回数据也不能抛出异常。

Future接口

package java.util.concurrent;

/**
 * 表示异步执行的结果，有三个功能：
 * 1.获取异步执行任务的结果
 * 2.查看异步任务的执行状态（取消或终止）
 * 3.取消异步任务
 */
public interface Future<V> 

    /**
     * 尝试取消任务，如果任务已经完成或已经取消，则取消失败。
     * 1.如果任务没未被启动，则该任务不会被运行；
     * 2.如果任务已经被启动，参数mayInterruptIfRunning决定是否执行当前任务
     *   的线程是否应该被中断，这只是作为一种终止任务的尝试。
     * 执行这个方法之后，以后的isDone方法调用都会返回true。
     * 如果这个方法返回true，以后的isCancelled方法调用都会返回true。
     */
    boolean cancel(boolean mayInterruptIfRunning);

    /**
     * cancel()调用返回true之后，这个方法会返回true。
     */
    boolean isCancelled();

    /**
     * 如果当前任务执行成功，或者被取消，或者抛出异常，则返回true
     */
    boolean isDone();

    /**
     * 阻塞直到任务完成，并返回任务执行结果。
     * 当异步任务被取消，或抛出异常，get()方法会抛出相应的异常
     */
    V get() throws InterruptedException, ExecutionException;

    /**
     * 阻塞一定时间等待任务完成，并返回任务执行结果，超过时间未返回结果会抛出异常
     */
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;

• cancel():cancel()方法用来取消异步任务的执行。如果异步任务已经完成或者已经被取消，或者由于某些原因不能取消，则会返回false。如果任务还没有被执行，则会返回true并且异步任务不会被执行。如果任务已经开始执行了但是还没有执行完成，若mayInterruptIfRunning为true，则会立即中断执行任务的线程并返回true，若mayInterruptIfRunning为false，则会返回true且不会中断任务执行线程。
• isCanceled():判断任务是否被取消，如果任务在结束(正常执行结束或者执行异常结束)前被取消则返回true，否则返回false。
• isDone():判断任务是否已经完成，如果完成则返回true，否则返回false。需要注意的是：任务执行过程中发生异常、任务被取消也属于任务已完成，也会返回true。
• get():获取任务执行结果，如果任务还没完成则会阻塞等待直到任务执行完成。如果任务被取消则会抛出CancellationException异常，如果任务执行过程发生异常则会抛出ExecutionException异常，如果阻塞等待过程中被中断则会抛出InterruptedException异常。
• get(long timeout,Timeunit unit):带超时时间的get()版本，如果阻塞等待过程中超时则会抛出TimeoutException异常。

FutureTask

Future只是一个接口，不能直接用来创建对象，FutureTask是Future的实现类，
FutureTask的继承图如下:

 

public class FutureTask<V> implements RunnableFuture<V> 
 
   /**
     * 当前任务的运行状态。
     *
     * 可能存在的状态转换
     * 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;  //任务正常完成，结果被set
    private static final int EXCEPTIONAL  = 3;  //任务抛出异常
    private static final int CANCELLED    = 4;  //任务已被取消
    private static final int INTERRUPTING = 5;  //线程中断状态被设置ture，但线程未响应中断
    private static final int INTERRUPTED  = 6;  //线程已被中断

    //将要执行的任务
    private Callable<V> callable;
    //用于get()返回的结果，也可能是用于get()方法抛出的异常
    private Object outcome; // non-volatile, protected by state reads/writes
    //执行callable的线程，调用FutureTask.run()方法通过CAS设置
    private volatile Thread runner;
    //栈结构的等待队列，该节点是栈中的最顶层节点。
    private volatile WaitNode waiters;
    ....

为了后面更好的分析FutureTask的实现，这里有必要解释下各个状态。

• NEW:表示是个新的任务或者还没被执行完的任务。这是初始状态。
• COMPLETING:任务已经执行完成或者执行任务的时候发生异常，但是任务执行结果或者异常原因还没有保存到outcome字段(outcome字段用来保存任务执行结果，如果发生异常，则用来保存异常原因)的时候，状态会从NEW变更到COMPLETING。但是这个状态会时间会比较短，属于中间状态。
• NORMAL:任务已经执行完成并且任务执行结果已经保存到outcome字段，状态会从COMPLETING转换到NORMAL。这是一个最终态。
• EXCEPTIONAL:任务执行发生异常并且异常原因已经保存到outcome字段中后，状态会从COMPLETING转换到EXCEPTIONAL。这是一个最终态。
• CANCELLED:任务还没开始执行或者已经开始执行但是还没有执行完成的时候，用户调用了cancel(false)方法取消任务且不中断任务执行线程，这个时候状态会从NEW转化为CANCELLED状态。这是一个最终态。
• INTERRUPTING: 任务还没开始执行或者已经执行但是还没有执行完成的时候，用户调用了cancel(true)方法取消任务并且要中断任务执行线程但是还没有中断任务执行线程之前，状态会从NEW转化为INTERRUPTING。这是一个中间状态。
• INTERRUPTED:调用interrupt()中断任务执行线程之后状态会从INTERRUPTING转换到INTERRUPTED。这是一个最终态。

有一点需要注意的是，所有值大于COMPLETING的状态都表示任务已经执行完成(任务正常执行完成，任务执行异常或者任务被取消)。

各个状态之间的可能转换关系如下图所示:

RunnableFuture 接口

public interface RunnableFuture<V> extends Runnable, Future<V> 
    void run();

RunnableFuture 接口基础了Runnable和Future接口。

FutureTask的run方法

public void run() 
    // 1. 状态如果不是NEW，说明任务或者已经执行过，或者已经被取消，直接返
    // 2. 状态如果是NEW，则尝试把当前执行线程保存在runner字段中
    // 如果赋值失败则直接返回,保证callable任务只被运行一次
        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 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);
        
    

run()方法首先会

1. 判断当前任务的state是否等于NEW,如果不为NEW则说明任务或者已经执行过，或者已经被取消，直接返回。
2. 如果状态为NEW则接着会通过unsafe类把任务执行线程引用CAS的保存在runner字段中，如果保存失败，则直接返回。
3. 执行任务。
4. 如果任务执行发生异常，则调用setException()方法保存异常信息。setException()方法如下：

protected void setException(Throwable t) 
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) 
            outcome = t;
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            finishCompletion();
        
    

在setException()方法中

• 首先会CAS的把当前的状态从NEW变更为COMPLETING状态。
• 把异常原因保存在outcome字段中，outcome字段用来保存任务执行结果或者异常原因。
• CAS的把当前任务状态从COMPLETING变更为EXCEPTIONAL。这个状态转换对应着上图中的二。
• 调用finishCompletion()。关于这个方法后面在分析。

如果任务成功执行则调用set()方法设置执行结果，该方法实现如下:

protected void set(V v) 
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) 
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        
    

这个方法跟上面分析的setException()差不多，

• 首先会CAS的把当前的状态从NEW变更为COMPLETING状态。
• 把任务执行结果保存在outcome字段中。
• CAS的把当前任务状态从COMPLETING变更为NORMAL。这个状态转换对应着上图中的一。
• 调用finishCompletion()。

发起任务线程跟执行任务线程通常情况下都不会是同一个线程，在任务执行线程执行任务的时候，任务发起线程可以查看任务执行状态、获取任务执行结果、取消任务等等操作，接下来分析下这些操作。

get()

任务发起线程可以调用get()方法来获取任务执行结果，如果此时任务已经执行完毕则会直接返回任务结果，如果任务还没执行完毕，则调用方会阻塞直到任务执行结束返回结果为止。get()方法实现如下:

public V get() throws InterruptedException, ExecutionException 
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    

get()方法实现比较简单，会

1. 判断任务当前的state <= COMPLETING是否成立。前面分析过，COMPLETING状态是任务是否执行完成的临界状态。
2. 如果成立，表明任务还没有结束(这里的结束包括任务正常执行完毕，任务执行异常，任务被取消)，则会调用awaitDone()进行阻塞等待。
3. 如果不成立表明任务已经结束，调用report()返回结果。

awaitDone()

当调用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 (;;) 
           // 判断阻塞线程是否被中断,如果被中断则在等待队
           // 列中删除该节点并抛出InterruptedException异常
            if (Thread.interrupted()) 
                removeWaiter(q);
                throw new InterruptedException();
            
           // 获取当前状态，如果状态大于COMPLETING
           // 说明任务已经结束(要么正常结束，要么异常结束，要么被取消)
           // 则把thread显示置空，并返回结果
            int s = state;
            if (s > COMPLETING) 
                if (q != null)
                    q.thread = null;
                return s;
            
           // 如果状态处于中间状态COMPLETING
           // 表示任务已经结束但是任务执行线程还没来得及给outcome赋值
           // 这个时候让出执行权让其他线程优先执行
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
           //如果等待节点为空，则构造一个等待节点
            else if (q == null)
                q = new WaitNode();
           //如果还没有入队列，则把当前节点加入waiters首节点并替换原来waiters
            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);
        
    

awaitDone()中有个死循环，每一次循环都会

1. 判断调用get()的线程是否被其他线程中断，如果是的话则在等待队列中删除对应节点然后抛出InterruptedException异常。
2. 获取任务当前状态，如果当前任务状态大于COMPLETING则表示任务执行完成，则把thread字段置null并返回结果。
3. 如果任务处于COMPLETING状态，则表示任务已经处理完成(正常执行完成或者执行出现异常)，但是执行结果或者异常原因还没有保存到outcome字段中。这个时候调用线程让出执行权让其他线程优先执行。
4. 如果等待节点为空，则构造一个等待节点WaitNode。
5. 如果第四步中新建的节点还没如队列，则CAS的把该节点加入waiters队列的首节点。
6. 阻塞等待。

假设当前state=NEW且waiters为NULL,也就是说还没有任何一个线程调用get()获取执行结果，这个时候有两个线程threadA和threadB先后调用get()来获取执行结果。再假设这两个线程在加入阻塞队列进行阻塞等待之前任务都没有执行完成且threadA和threadB都没有被中断的情况下(因为如果threadA和threadB在进行阻塞等待结果之前任务就执行完成或线程本身被中断的话，awaitDone()就执行结束返回了)，执行过程是这样的，以threadA为例:

1. 第一轮for循环，执行的逻辑是q == null,所以这时候会新建一个节点q。第一轮循环结束。
2. 第二轮for循环，执行的逻辑是!queue，这个时候会把第一轮循环中生成的节点的netx指针指向waiters，然后CAS的把节点q替换waiters。也就是把新生成的节点添加到waiters链表的首节点。如果替换成功，queued=true。第二轮循环结束。
3. 第三轮for循环，进行阻塞等待。要么阻塞特定时间，要么一直阻塞知道被其他线程唤醒。

在threadA和threadB都阻塞等待之后的waiters结果如图

cancel(boolean)

用户可以调用cancel(boolean)方法取消任务的执行，cancel()实现如下:

public boolean cancel(boolean mayInterruptIfRunning) 
        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 
            finishCompletion();
        
        return true;
    

• 根据mayInterruptIfRunning是否为true，CAS设置状态为INTERRUPTING或CANCELLED，设置成功，继续第二步，否则返回false
• 如果mayInterruptIfRunning为true，调用runner.interupt()，设置状态为INTERRUPTED
• 唤醒所有在get()方法等待的线程

finishCompletion()

根据前面的分析，不管是任务执行异常还是任务正常执行完毕，或者取消任务，最后都会调用finishCompletion()方法，该方法实现如下:

private void finishCompletion() 
        // assert state > COMPLETING;
        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;
            
        

        done();

        callable = null;        // to reduce footprint
    

这个方法的实现比较简单，依次遍历waiters链表，唤醒节点中的线程，然后把callable置空。

被唤醒的线程会各自从awaitDone()方法中的LockSupport.park*()阻塞中返回，然后会进行新一轮的循环。在新一轮的循环中会返回执行结果(或者更确切的说是返回任务的状态)。

report()

report()方法用在get()方法中，作用是把不同的任务状态映射成任务执行结果。实现如下：

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的状态为NORMAL，说明任务正确执行完成，直接返回计算后的值。
如果state的状态大于等于CANCELLED，说明任务被成功取消执行、或响应中断，直接返回CancellationException异常
否则返回ExecutionException异常。

get(long,TimeUnit)

带超时等待的获取任务结果，实现如下:

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);
    

跟get()不同点在于get(long,TimeUnit)会在awaitDone()超时返回之后抛出TimeoutException异常。

isCancelled()和isDone()

这两个方法分别用来判断任务是否被取消和任务是否执行完成，实现都比较简单，代码如下:

public boolean isCancelled() 
        return state >= CANCELLED;


public boolean isDone() 
        return state != NEW;

总结下，其实FutureTask的实现还是比较简单的，当用户实现Callable()接口定义好任务之后，把任务交给其他线程进行执行。FutureTask内部维护一个任务状态，任何操作都是围绕着这个状态进行，并随时更新任务状态。任务发起者调用get*()获取执行结果的时候，如果任务还没有执行完毕，则会把自己放入阻塞队列中然后进行阻塞等待。当任务执行完成之后，任务执行线程会依次唤醒阻塞等待的线程。调用cancel()取消任务的时候也只是简单的修改任务状态，如果需要中断任务执行线程的话则调用Thread.interrupt()中断任务执行线程。

有个值得关注的问题就是当任务还在执行的时候用户调用cancel(true)方法能否真正让任务停止执行呢？
在前面的分析中我们直到，当调用cancel(true)方法的时候，实际执行还是Thread.interrupt()方法，而interrupt()方法只是设置中断标志位，如果被中断的线程处于sleep()、wait()或者join()逻辑中则会抛出InterruptedException异常。

因此结论是:cancel(true)并不一定能够停止正在执行的异步任务。

参考：https://www.cnblogs.com/linghu-java/p/8991824.html