Java中的Future详解

1.什么是Future

线程的使用方式有两种：继承Thread类，覆盖其run方法；创建实现Runnable方法的实例，通过Thread类构造函数传入。这里存在两个问题：

• 无法知道线程的执行状态(join方法可以阻塞当前线程等待另一个线程执行完成)；

• 线程没有返回值。

Java在JDK1.5引入了Future与Callable，可以获取线程的执行状态以及获得线程的返回值。

2.Future的继承结构

通过IEDA(快捷键Ctrl+Shift+Alt+U)可以很方便的看到Future的继承结构：

通过继承图可以看到：

• Future是一个接口，可以cancel线程、获取线程执行状态和执行结果
• RunnableFuture也是个接口，继承自Future与Runnable
• FutureTask是RunnableFuture的实现类，提供了一些可以获取线程执行状态的方法

FutureTask类间接实现了Future与Runnable，所以FutureTask既可以被线程执行，也可以作为Future得到线程的执行状态和返回值。

3.FutureTask的使用

场景：我们需要计算A+B，但是A需要2s才能准备好，B需要3秒。

public static void main(String[] args) throws ExecutionException, InterruptedException {

    long startTime = System.currentTimeMillis();

    FutureTask<Integer> futureA = new FutureTask<>(new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            Thread.sleep(3000);
            return 1;
        }
    });

    FutureTask<Integer> futureB = new FutureTask<>(new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            Thread.sleep(2000);
            return 1;
        }
    });

    new Thread(futureA).start();
    new Thread(futureB).start();

    int A = futureA.get();
    int B = futureB.get();

    System.out.println("A+B=" + A+B);

    long endTime = System.currentTimeMillis();

    System.out.println("总共花费：" + (endTime - startTime) + "s");


}

/**
A+B=11
总共花费：3005s
*/

这里使用了两个线程分别去准备A和B，这里总共花费了3s左右，而不是5s。因为get方法会阻塞直到线程执行完毕，返回返回值，所以总的执行时间是max(A所需时间, B所需时间)。

4.FutureTask源码剖析

1.基础

1.FutureTask的属性

public class FutureTask<V> implements RunnableFuture<V> {
   /** 
     * 任务执行的状态，初始值是NEW
     * 可能的状态变化如下:
     * 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 underlying callable; nulled out after running */
    // 任务需要封装成Callable
    private Callable<V> callable;
    /** 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;
    /** Treiber stack of waiting threads */
    // 等待任务结果的队列
    private volatile WaitNode waiters;
    ****

2.线程等待队列WaitNode

// 一个存放等待线程的单链表
static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
}

3.Unsafe工具类

// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;   // 标志位在对象中的偏移
private static final long runnerOffset;  // ...
private static final long waitersOffset;
static {
    try {
        UNSAFE = sun.misc.Unsafe.getUnsafe();
        Class<?> k = FutureTask.class;
        stateOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("state"));
        runnerOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("runner"));
        waitersOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("waiters"));
    } catch (Exception e) {
        throw new Error(e);
    }
}

这里主要是得到对象属性如state相对于对象起始物理内存地址的偏移，后续通过CAS进行操作。

4.构造方法

public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

构造方法是为了传入带返回值的任务(Callable)，第二个构造方法需要构造构一个实现了Callable对象(实际上是RunnableAdapter)。

如果使用的是第二个构造函数，那么任务的结果可以通过result引用直接访问。

2.核心

1.run方法

public void run() {
    // 任务状态不是NEW或CAS设置任务的runner线程失败（不是NULL，说明已经启动过了），直接结束
    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);   // 保存任务执行过程中的异常
            }
            // 任务正常执行完成，将结果保存到FutureTask
            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);
    }
}

其中的setException方法会将异常作为结果，并且修改任务的执行状态：

//发生异常时设置state和outcome：异常发生时，NEW->COMPLETING->EXCEPTIONAL
protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;   // 保存异常作为任务结果
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);    // 设置任务状态：EXCEPTIONAL
        finishCompletion();// 唤醒get()方法阻塞的线程
    }
}

其中setResult方法会保存任务结果，并且设置任务状态：

//任务正常完成时，设置state和outcome:NEW->COMPLETING->NORMAL
protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
         outcome = v;
         UNSAFE.putOrderedInt(this, stateOffset, NORMAL); 
         finishCompletion(); // 唤醒get方法阻塞的线程
    }
}

任务异常/正常结束后都会用finishCompletion方法唤醒get方法阻塞的线程：

//移除并唤醒所有等待的线程，调用done，并清空callable
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); //唤醒线程
                }
                //接下来的这几句代码是将当前节点剥离出队列，然后将q指向下一个等待节点。被剥离的节点由于thread和next都为null，所以会被GC回收。
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }

    done(); //这个是空的方法，子类可以覆盖，实现回调的功能。
    callable = null;        // to reduce footprint
}

UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)这行代码卡了好久，是因为对CAS理解存在问题，意思如下：

1.首先q = waiters,都指向等待队列
2.然后CAS判断waiters是否等于q，等于的话就令waiters=null;否则继续重复1.2.
3.然后通过q来访问队列，逐个唤醒线程
waiters设置为null后，相当于对队列进行了截断：waiters和q（逐个移除唤醒）。后边可能还会有线程加入waiters，所以外面设置了一个for循环来保证waiters全部被唤醒。

这里使用CAS的原因应该是因为等待队列采用的是头插法，通过CAS保证多线程安全。凡是涉及到多线程同步的地方都不好理解。

2.get()方法

get方法有两个，一个是有超时时间设置，另一个没有超时时间：

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    // 任务状态<=COMPLETING,说明任务还未执行完毕
    if (s <= COMPLETING)
        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;
    // 任务状态<=COMPLETING,说明任务还未执行完毕
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

get方法的关键在于awaitDone方法，线程在不断等待任务执行完成：

//等待完成，可能是是中断、异常、正常完成，timed:true，考虑等待时长，false:不考虑等待时长
private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        /**
         *  有优先级顺序
         *  1、如果线程已中断，则直接将当前节点q从waiters中移出
         *  2、如果state已经是最终状态了，则直接返回state
         *  3、如果state是中间状态(COMPLETING),意味很快将变更过成最终状态，让出cpu时间片即可
         *  4、如果发现尚未有节点，则创建节点
         *  5、如果当前节点尚未入队，则将当前节点放到waiters中的首节点，并替换旧的waiters
         *  6、线程被阻塞指定时间后再唤醒
         *  7、线程一直被阻塞直到被其他线程唤醒
         *
         */
        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)
            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);
    }
}

removeWaiter移除等待队列中的节点：

private void removeWaiter(WaitNode node) {
    if (node != null) {
        node.thread = null;     // 将移除的节点的thread＝null, 为移除做标示
        retry:
        for (;;) {          // restart on removeWaiter race
            for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                s = q.next;
                //通过 thread 判断当前 q 是否是需要移除的 q节点，因为我们刚才标示过了
                if (q.thread != null)
                    pred = q;   //当不是我们要移除的节点，就往下走
                else if (pred != null) {
                    //当p.thread==null时，到这里。下面这句话，相当于把q从队列移除。
                    pred.next = s;
                    //pred.thread == null 这种情况是在多线程进行并发 removeWaiter 时产生的
                        //此时正好移除节点 node 和 pred, 所以loop跳到retry, 从新进行这个过程。想象一下，如果在并发的情况下，其他线程把pred的线程置为空了。那说明这个链表不应该包含pred了。所以我们需要跳到retry从新开始。
                    if (pred.thread == null) // check for race
                        continue retry;
                }
                //到这步说明p.thread==null 并且 pred==null。说明node是头结点。
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                      q, s))
                    continue retry;
            }
            break;
        }
    }
}

最后在get方法中调用report(s)，根据状态s的不同进行返回结果或抛出异常。

private V report(int s) throws ExecutionException {
    Object x = outcome;     //之前我们set的时候，已经设置过这个值了。所以直接用。
    if (s == NORMAL)   //正常执行结束，返回结果
        return (V)x;
    if (s >= CANCELLED)   //被取消或中断了，就抛异常。
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

3.难点

awaitDone和removeWaiter方法涉及了多线程并发，不太容易理解。