FutureTask原理解析

FutureTask原理解析 引言

我们都知道Runable接口，它非常简单。但是它有一个问题——无法获取执行结果，以及一旦可以获取执行结果，什么时候可以获取执行结果？

public interface Runnable {
    public abstract void run();
}

FutureTask就是为了解决这个问题的，将Runnable包装为FutureTask之后，就可以get获取任务执行结果，如果任务没有执行完，那么当前线程就会阻塞。
FutureTask确实非常好用，但我一直以来都比较好奇，将Runnable包装为FutureTask，为何就能实现执行结果的自动获取？或者换句话说，FutureTask的原理究竟是什么？

概述

我们先来概述一下整体的结构以及整体的设计。

FutureTask实现了RunnaleFuture接口，而后者则是继承自两个接口Future和Runnable，相当于两个接口的混合（接口支持多继承）。
事实上，Future接口也就定义了FutureTask类特性的交互协议。同样，它也非常简单。那么如何实现阻塞等待执行结果呢？

public interface Future {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

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

考虑到多线程的环境——可能会有多个线程同时获取结果，一起阻塞，所以很自然地，我们需要将这些线程阻塞等待在一个列表之上，当任务执行完成之后，唤醒这些线程获取结果。
这也是FutureTask整体的设计思想，接下来，我们通过源码解析，探究它如何通过CAS的方式实现任务阻塞等待，以及维护等待队列。

源码解析 回调

Runnable是不返回结果的，所以首先，他会将Runnalbe执行和它关联的执行结果Result包装为带返回结果的Callable。这些都非常简单。

public interface Callable {
    V call() throws Exception;
}
static final class RunnableAdapter implements Callable {
    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构造函数，将二者包装为callable
public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

状态

FutureTask中设置了多种状态变量，用于标志任务执行的状态。理解源代码，首先需要理解这些状态的变化协议。

state有四种可能的状态转换：

1. NEW -> COMPLETING -> NORMAL
2. NEW -> COMPLETING -> EXCEPTIONAL
3. NEW -> CANCELLED
4. NEW -> INTERRUPTING -> INTERRUPTED

其中

• NEW为初始状态，任务执行过程也为该状态。
• COMPLETING 为中间状态，表示任务已经执行完，有线程正在调用set设置结果
• INTERRUPTING 为中间状态，表示正在中断中
• NORMAL，EXCEPTIONAL，CANCELLED，INTERRUPTED则为终结状态，分别表示正常结束，异常结束，任务取消，被中断

源码注释

有了上述知识，FutureTask的源码应该算比较简单，这里提供了核心代码的注释，从以下两个方面切入即可：

• run，任务执行，执行完调用set，set会调用finishCompletion唤醒所有等待线程
• get，获取结果，如果没有完成则阻塞当前线程插入队列

其次，FutureTask广泛使用了CAS，例如：

UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread())

compareAndSwapObject(对象，字段偏移量，期待的值，新的值)，它会判断对象中的字段是否为期待值，如果是，则设置为新的值，并且返回true，否则返回false。注意CAS底层为native实现的原子方法。

public class FutureTask implements RunnableFuture {


    
    private Callable callable;
    
    private Object outcome; // non-volatile, protected by state reads/writes
    
    private volatile Thread runner;
    
    private volatile WaitNode waiters;



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

    
    public void run() {
        // 状态不为NEW或者当前runner持有线程，则直接返回。只有一个线程能执行run方法
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable 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);
        }
    }

 
    
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            // 阻塞当前线程
            s = awaitDone(false, 0L);
        return report(s);
    }


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

            int s = state;
            // 如果大于COMPLETING，表明task执行完成，直接返回结果
            if (s > COMPLETING) {
                if (q != null)
                    q.thread = null;
                return s;
            }
            // 如果等于COMPLETING，说明已经有线程正在执行set方法，让出执行权限
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
            // 如果状态为NEW（小于COMPLETING），则通过两次循环入队等待
            else if (q == null)
                // 封装当前线程为等待节点
                q = new WaitNode();
            else if (!queued)
                // q.next = waiters，CAS修改waiters，之后queued=true，当前线程循环不会再进入
                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);
        }
    }


    
    protected void done() { }

    
    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            // CAS设置状态
            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();
        }
    }
    
    private void finishCompletion() {
        // assert state > COMPLETING;
        for (WaitNode q; (q = waiters) != null;) {
            // 进入if之后CAS将等待列表头部节点置为null，保证只唤醒一次
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (;;) {
                    Thread t = q.thread;
                    // 唤醒当前节点上的线程
                    if (t != null) {
                        q.thread = null;
                        LockSupport.unpark(t);
                    }
                    // next
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }
        // 执行回调
        done();

        callable = null;        // to reduce footprint
    }

   
    public boolean cancel(boolean mayInterruptIfRunning) {
        // 如果当前状态为NEW则CAS修改为中断中或取消
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            // 状态不为NEW或CAS失败，不为NEW表示已经处于完成、中断、取消状态，这些状态不能被取消，直接返回false
            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;
    }

  
    
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }



    // 设置内存偏移量
    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);
        }
    }

}

参考

https://blog.wangqi.love/articles/Java/Callable%E3%80%81Future%E3%80%81FutureTask%E5%8E%9F%E7%90%86.html