异步任务——FutureTask

Runnable、Callable、Future、FutureTask

和Java异步打交道就不能回避掉Runnable，Callable，Future，FutureTask等类，首先来介绍下这几个类的区别。

Runnable

Runnable接口是我们最熟悉的，它只有一个run函数。然后使用某个线程去执行该runnable即可实现多线程，Thread类在调用start()函数后就是执行的是Runnable的run()函数。Runnable最大的缺点在于run函数没有返回值。

Callable

Callable接口和Runnable接口类似，它有一个call函数。使用某个线程执行Callable接口实质就是执行其call函数。call方法和run方法最大的区别就是call方法有返回值：

public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

Callable 接口有返回值，泛型 V 就是要返回的结果类型，可以返回子任务的执行结果。

Future接口

Future 接口表示异步计算的结果，通过 Future 接口提供的方法，可以很方便的查询异步计算任务是否执行完成，获取异步计算的结果，取消未执行的异步任务，或者中断异步任务的执行，接口定义如下：

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
}

cancel(boolean mayInterruptIfRunning)：取消子任务的执行，如果这个子任务已经执行结束，或者已经被取消，或者不能被取消，这个方法就会执行失败并返回false；如果子任务还没有开始执行，那么子任务会被取消，不会再被执行；如果子任务已经开始执行了，但是还没有执行结束，根据mayInterruptIfRunning的值，如果mayInterruptIfRunning = true，那么会中断执行任务的线程，然后返回true，如果参数为false，会返回true，不会中断执行任务的线程。
需要注意，这个方法执行结束，返回结果之后，再调用isDone()会返回true。

isCancelled()：判断任务是否被取消，如果任务执行结束（正常执行结束和发生异常结束，都算执行结束）前被取消，也就是调用了cancel()方法，并且cancel()返回true，则该方法返回true，否则返回false。

isDone()：判断任务是否执行结束，正常执行结束，或者发生异常结束，或者被取消，都属于结束，该方法都会返回true.

V get()：获取结果，如果这个计算任务还没有执行结束，该调用线程会进入阻塞状态。如果计算任务已经被取消，调用get()会抛出CancellationException，如果计算过程中抛出异常，该方法会抛出ExecutionException，如果当前线程在阻塞等待的时候被中断了，该方法会抛出InterruptedException。

V get(long timeout, TimeUnit unit)：带超时限制的get()，等待超时之后，该方法会抛出TimeoutException，如果子任务执行结束了，但是超时时间还没有到，这个方法也会返回结果。

FutureTask

Future只是一个接口，在实际使用过程中，诸如ThreadPoolExecutor返回的都是一个FutureTask实例。

1
2
3

public class FutureTask<V> implements RunnableFuture<V> 

public interface RunnableFuture<V> extends Runnable, Future<V>

FutureTask可以像Runnable一下，封装异步任务，然后提交给Thread或线程池执行，然后获取任务执行结果。原因在于 FutureTask 实现了 RunnableFuture 接口，RunnableFuture是什么呢，其实就是 Runnable 和 Future 的结合，实现了 Runnable 和 Future 两个接口。

FutureTask的成员变量

public class FutureTask<V> implements RunnableFuture<V> {
     /*
     * FutureTask中定义了一个state变量，用于记录任务执行的相关状态 ，状态的变化过程如下
     * NEW -> COMPLETING -> NORMAL
     * NEW -> COMPLETING -> EXCEPTIONAL
     * NEW -> CANCELLED
     * NEW -> INTERRUPTING -> INTERRUPTED
     */
    private volatile int state;
    //主流程状态
    private static final int NEW = 0; //当FutureTask实例刚刚创建到callbale的call方法执行完成前，处于此状态
    private static final int COMPLETING  = 1; //callable的call方法执行完成或出现异常时，首先进行此状态
    private static final int NORMAL    = 2;//callable的call方法正常结束时，进入此状态，将outcom设置为正常结果
    private static final int EXCEPTIONAL = 3;//callable的call方法异常结束时，进入此状态，将outcome设置为抛出的异常
    //取消任务执行时可能处于的状态
    private static final int CANCELLED= 4;// FutureTask任务尚未执行，即还在任务队列的时候，调用了cancel方法，进入此状态
    private static final int INTERRUPTING = 5;// FutureTask的run方法已经在执行，收到中断信号，进入此状态
    private static final int INTERRUPTED  = 6;// 任务成功中断后，进入此状态
    
    private Callable<V> callable;//需要执行的任务，提示：如果提交的是Runnable对象，会先转换为Callable对象，这是构造方法参数
    private Object outcome; //任务运行的结果
    private volatile Thread runner;//执行此任务的线程
  
    //等待该FutureTask的线程链表，对于同一个FutureTask，如果多个线程调用了get方法，对应的线程都会加入到waiters链表中，同时当FutureTask执行完成后，也会告知所有waiters中的线程
    private volatile WaitNode waiters;
    ......
}

private volatile int state，state用来标识当前任务的运行状态。FutureTask的所有方法都是围绕这个状态进行的，需要注意，这个值用volatile（易变的）来标记，如果有多个子线程在执行FutureTask，那么它们看到的都会是同一个state，有如下几个值：

NEW：表示这是一个新的任务，或者还没有执行完的任务，是初始状态。
COMPLETING：表示任务执行结束（正常执行结束，或者发生异常结束），但是还没有将结果保存到outcome中。是一个中间状态。
NORMAL：表示任务正常执行结束，并且已经把执行结果保存到outcome字段中。是一个最终状态。
EXCEPTIONAL：表示任务发生异常结束，异常信息已经保存到outcome中，这是一个最终状态。
CANCELLED：任务在新建之后，执行结束之前被取消了，但是不要求中断正在执行的线程，也就是调用了cancel(false)，任务就是CANCELLED状态，这时任务状态变化是NEW -> CANCELLED。
INTERRUPTING：任务在新建之后，执行结束之前被取消了，并要求中断线程的执行，也就是调用了cancel(true)，这时任务状态就是INTERRUPTING。这是一个中间状态。
INTERRUPTED：调用cancel(true)取消异步任务，会调用interrupt()中断线程的执行，然后状态会从INTERRUPTING变到INTERRUPTED。

状态变化有如下4种情况：
NEW -> COMPLETING -> NORMAL ————————————— 正常执行结束的流程
NEW -> COMPLETING -> EXCEPTIONAL ———————执行过程中出现异常的流程
NEW -> CANCELLED ——————————————-被取消，即调用了cancel(false)
NEW -> INTERRUPTING -> INTERRUPTED ————-被中断，即调用了cancel(true)

FutureTask的构造方法

//接受Callable对象作为参数
public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;    
}
//接受Runnable对象作为参数
public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);//将Runnable转为Callable对象
    this.state = NEW;     
}

//callable方法，将Runnable转为一个Callable对象，包装设计模式
public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}
 
//RunnableAdapter是Executors的一个内部类，实现了Callable接口
static final class RunnableAdapter<T> implements Callable<T> {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

可以看到，构造FutureTask时，无论传入的是Runnable还是Callable，最终都实现了Callable接口。

FutureTask的执行过程

run

public void run() {
    //保证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 = null;
        int s = state;
        //判断该任务是否正在响应中断，如果中断没有完成，则等待中断操作完成
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

如果状态不为new或者运行线程runner失败，说明当前任务已经被其他线程启动或者已经被执行过，直接返回false
调用call方法执行核心任务逻辑。如果调用成功则执行set(result)方法，将state状态设置成NORMAL。如果调用失败抛出异常则执行setException(ex)方法，将state状态设置成EXCEPTIONAL，唤醒所有在get()方法上等待的线程
如果当前状态为INTERRUPTING(步骤2已CAS失败)，则一直调用Thread.yield()直至状态不为INTERRUPTING

set、setException方法

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

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

两个方法的逻辑基本一致，先通过CAS操作将状态从NEW置为COMPLETING，然后再将最终状态分别置为NORMAL或者EXCEPTIONAL，最后再调用finishCompletion方法。

get

在finishCompletion方法中，FutureTask会通知waiters链表中的每一个等待线程，那么这些线程是怎么被加入到waiters链表中的呢？上文已经讲过，当在一个线程中调用了get方法，该线程就会被加入到waiters链表中。所以接下来看下get方法：

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

其方法主要是调用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 (;;) {
        //如果该线程执行interrupt()方法，则从队列中移除该节点，并抛出异常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        int s = state;
        //如果state状态大于COMPLETING 则说明任务执行完成，或取消
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        //如果state=COMPLETING，则使用yield，因为此状态的时间特别短，通过yield比挂起响应更快。
        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);
        //如果需要阻塞指定时间，则使用LockSupport.parkNanos阻塞指定时间
        //如果到指定时间还没执行完，则从队列中移除该节点，并返回当前状态
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
                }
                LockSupport.parkNanos(this, nanos);
            }
            //阻塞当前线程
            else
                LockSupport.park(this);
        }
}

整个方法的大致逻辑主要分为以下几步：

如果当前状态值大于COMPLETING，说明已经执行完成或者取消，直接返回

如果state=COMPLETING，则使用yield，因为此状态的时间特别短，通过yield比挂起响应更快

如果当前线程是首次进入循环，为当前线程创建wait节点加入到waiters链表中

根据是否定时将当前线程挂起（LockSupport.parkNanos， LockSupport.park）来阻塞当前线程，直到超时或者线程被finishCompletion方法唤醒

当线程挂起超时或者被唤醒后，重新循环执行上述逻辑

get方法是FutureTask中的关键方法，了解了get方法逻辑也就了解为什么当调用get方法时线程会被阻塞直到任务运行完成。

finishCompletion

private void finishCompletion() {
    for (WaitNode q; (q = waiters) != null;) {
        //通过CAS把栈顶的元素置为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
}

finishCompletion的逻辑也比较简单：

遍历waiters链表，取出每一个节点：每个节点都代表一个正在等待该FutureTask结果（即调用过get方法）的线程

通过 LockSupport.unpark(t)唤醒每一个节点，通知每个线程，该任务执行完成

cancel

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()方法等待的线程

其他方法

public boolean isCancelled() {
    return state >= CANCELLED;
}

public boolean isDone() {
    return state != NEW;
}

这两个方法其实就是判断state，没有过多的步骤。

参考

https://www.jianshu.com/p/55221d045f39

https://yq.aliyun.com/articles/128032

https://blog.csdn.net/chenliguan/article/details/54345993

http://www.cnblogs.com/dolphin0520/p/3949310.html

http://www.threadworld.cn/archives/39.html