通俗易懂的JUC源码剖析-FutureTask

Posted 2021-10-05 小强大人

前言

FutureTask是Future接口的常见实现类，它用来描述异步的执行结果。它还实现了Runnable接口，换句话说，它可以被submit到线程池中运行，然后调用Future相关的接口获取执行结果。

实现原理

先来看类结构

public class FutureTask<V> implements RunnableFuture<V> {
}

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

public interface Future<V> {
    // 取消任务, mayInterruptIfRunning代表是否中断执行任务的线程
    boolean cancel(boolean mayInterruptIfRunning);
    // 任务是否被取消
    boolean isCancelled();
    // 任务是否完成
    boolean isDone();
    // 获取任务结果，如果任务已取消，抛出CancelledException
    V get() throws InterruptedException, ExecutionException;
    // 限时获取任务结果
    V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException;
}

可以看到，FutureTask实现了RunnableFuture接口，而RunnableFuture同时实现了Future和Runnable接口，也就是说它拥有这两个接口的特性，既可以submit到线程池运行，也可以get()获取任务执行结果。

我们来看看它是如何获取到任务执行结果的呢。

1.构造函数

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

这个构造函数在ExecutorService.submit(Callable callable)中就用到了：

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

2.run()

public void run() {
    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()里面会调用callabe.call()，然后保存返回结果（正常或异常）。来看下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();
    }
}

可以看到，这2个方法逻辑类似，用outcome来保存结果，然后执行任务完成后的相关逻辑。其中state的枚举值如下：

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;

3.get()

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    // 如果任务状态还未完成，等待
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    // 否则返回结果    
    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 (;;) {
        // 如果当前线程被中断，从队列中移除该节点，并抛出中断异常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        int s = state;
        // 如果任务已完成（正常/异常/已取消），则将结点线程置空，并返回状态值
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        // 如果任务执行完，但outcome还没赋值完，
        // 让出当前线程执行权（底层CPU并不保证），
        // 让其他线程先设置最终状态。
        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;
            }
            // 阻塞当前线程nanos时长
            LockSupport.parkNanos(this, nanos);
       }
        // 无限阻塞场景，等待执行完run()的其他线程唤醒
        else
            LockSupport.park(this);
    }
}

其中WaitNode是个单向链表：

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

report()方法如下：

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

我们可以看到，调用get()方法时，如果任务未完成，会调用LockSupport.park阻塞自己，那么什么时候唤醒呢？谁来唤醒呢？

当然是执行run()的线程，上面提到，set(result)和setException(t)都会执行finishCompletion()，唤醒的逻辑就在这个里面。

private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        // CAS将等待队列头部指针waiters置空（可能存在竞争）
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            // CAS成功，挨个唤醒在get()等待队列中的线程
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                // 从链表断开，帮助GC回收内存    
                q.next = null; // unlink to help gc
                q = next;
            }
            // CAS成功，且唤醒工作完成，可以退出循环，
            // 否则CAS失败需要继续重试
            break;
        }
    }
    // 抽象的钩子方法，默认为空，留给子类覆盖
    done();
    // 显示置空，帮助GC释放内存
    callable = null; // to reduce footprint
}

4.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
                // 修改最终状态，没有用putIntVolatile是因为方法入口已经CAS操作成功
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        // 唤醒在阻塞在get()等待队列的线程
        finishCompletion();
    }
    return true;
}

参考资料：
https://dayarch.top/p/java-fu...