正文

JDK1.8源码分析：Future和FutureTask-任务异步执行结果

toobug
toobugV管理员/1.44 W阅读/0评论
0326
此篇文章发布距今已超过1484天，您需要注意文章的内容或图片是否可用！

概述

  • 任务在Executor线程执行器当中是异步执行的，而有些任务是需要返回执行结果的，故在Executor派生接口ExecutorService接口中定义了带返回结果的提交方法submit，其中返回结果为Future接口的对象。

  • Future接口主要提供了异步返回任务执行结果，取消任务执行，获取任务执行状态的功能，接口定义如下：

    public interface Future<V> {
    // 取消任务执行
    // mayInterruptIfRunning用于控制如果任务正在执行，是否中断对应的执行线程来取消该任务
    // 成功cancel，则isCancelled和isDoned都返回true。
    boolean cancel(boolean mayInterruptIfRunning);

    // 任务是否取消
    boolean isCancelled();

    // 正常执行，被取消，异常退出都返回true
    boolean isDone();

    // 阻塞等待执行结果
    // CancellationException：任务被取消
    // ExecutionException：任务执行异常
    // InterruptedException：该等待结果线程被中断
    V get() throws InterruptedException, ExecutionException;

    // 阻塞等待执行结果指定时间，除了以上异常，
    // TimeoutException：等待超时
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

FutureTask

  • Future接口的主要实现类为FutureTask，FutureTask同时实现了Runnable和Future接口，故对应的对象实例可以作为任务提交到Executor线程执行器中执行，然后通过自身来获取任务执行结果或者取消任务执行：

  • 即FutureTask的对象实例被Executor线程执行器内部线程池的某个工作线程和调用get方法等待获取结果的应用主线程所共享，故Executor内部线程池的工作线程在执行完这个任务后，可以通知和唤醒调用get阻塞等待执行结果的应用主线程，应用主线程也可以取消该任务的执行，然后通知工作线程。

    // 被线程池ExecutorService和调用get方法等待节点的主线程共享
// 线程池的执行线程worker调用run方法；等待结果的线程调用get等待结果，
// get通过自旋和LockSupport.park来先休眠，后等待执行线程在执行完run方法，调用LockSupport.unpark来唤醒，
// 调用get的线程被唤醒后，由于存在自旋，故再次检查是否有结果了，由结果则可以返回了。
public class FutureTask<V> implements RunnableFuture<V> {
    ...
}

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

  • 在FutureTask中定义了volatile修饰的状态变量state来进行Executor中的工作线程和应用主线程之间的交互，即工作线程产生任务执行结果，通知应用主线程获取；应用主线程请求取消任务执行，通知工作线程停止该任务执行。在内部实现中通过将state与以下状态常量进行大小比较来获取任务执行情况，如是正常执行成功还是异常退出，被取消等。

    /*
 * 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 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;

提交任务到Executor线程执行器

  • 提交任务到Executor线程执行器，对应AbstractExecutorService的submit方法实现如下：在submit中创建了一个FutureTask对象来包装应用定义的Runnable接口实现类task，调用execute将该对象交给Executor线程执行器执行，然后返回该对象引用给应用主线程。

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

/**
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 */
public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    // 将应用代码中的task包装成一个FutureTask，然后交给线程池执行
    // 线程池调度一个线程来执行FutureTask的run，在FutureTask的run中调用task的run
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    // 返回这个包装了task的FutureTask，在应用代码中可以通过get来获取执行结果
    return ftask;
}

应用主线程调用get等待执行结果

  • 在FutureTask中的get方法实现如下：

    public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        // 休眠等待执行结果
        s = awaitDone(false, 0L);
    // 休眠返回了，调用report获取结果
    return report(s);
}

  • 任务状态state小于等于COMPLETING表示任务还没开始执行，则应用主线程调用awaitDone阻塞休眠，等待Executor的工作线程执行任务并通知唤醒该应用主线程。具体过程如下：

    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;
        // 在set中，将状态从COMPLETING变成了NORMAL，并调用finishCompletion唤醒等待线程集合的所有线程，
        // NORMAL大于COMPLETING，故可以返回了。
        // 另外其他不是正常执行而结束的情况，如CANCELED, INTERRUPTED, EXCEPTIONAL等也是在这里返回
        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)
            // 下一轮自旋到下一个else if (!queued)，然后入队waiters
            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
            // 被finishCompletion唤醒，自旋回到for循环重新往下执行
            LockSupport.park(this);
    }
}

  • 在FutureTask内部维护了一个单向链表waiters，用于存放当前等待该任务执行结果的线程，在任务执行完成时，遍历该链表，唤醒每个等待线程。

    /** Treiber stack of waiting threads */
private volatile WaitNode waiters;
static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    // thread为调用get方法的线程，通常为主线程
    WaitNode() { thread = Thread.currentThread(); }
}

Executor工作线程执行任务

  • Executor的工作线程执行该任务时，会调用该任务的run方法，即FutureTask的run方法，如下为FutureTask的run方法定义：首先检查任务状态state是否为NEW，是，即还没执行过也没有被取消等，则进行往下执行。执行完成之后，产生执行结果result，调用set方法来处理这个结果。

    public void run() {
    // 赋值runner，指向线程池中当前执行这个task的线程
    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 {
                // 应用代码的task的run方法执行
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            // 执行成功设置结果result，并通知get方法
            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);
    }
}

  • set方法的定义如下：将执行结果赋值给FutureTask的成员变量outcome，更新任务执行状态state为NORMAL，最后调用finishCompletion通知所有等待这个任务执行结果的线程。

    protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        // 修改状态为NORMAL，NORMAL大于COMPLETING
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        // 唤醒调用get方法等待结果的线程
        finishCompletion();
    }
}

  • finishCompletion的实现如下：遍历任务等待线程链表，使用LockSupport.unpart唤醒对应的线程，然后将该等待线程从链表中移除。

    private void finishCompletion() {
    // assert state > COMPLETING;
    // waiters为调用了get方法的线程集合，通常为主线程
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    // 唤醒等待线程，即调用了get方法的线程
                    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
}

  • 然后回到get方法，应用主线程从awaitDone阻塞返回后，通过report方法来检测执行状态并返回任务执行结果。

    public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        // 休眠等待执行结果
        s = awaitDone(false, 0L);
    // 休眠返回了，调用report获取结果
    return report(s);
}
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);
}

应用主线程取消任务

  • 在应用主线程中，可以通过调用FutureTask的cancel方法来取消该任务的执行，cancel方法的定义如下：主要是更新任务的状态state为INTERRUPTING或者CANCELLED，然后根据mayInterruptIfRunning来控制如果该任务已经在执行，是否中断对应的工作线程来中止该任务的执行，最后调用finishCompletion方法来唤醒等待这个任务执行结果的线程，避免该任务被取消后，这些线程还在阻塞等待结果。

    public boolean cancel(boolean mayInterruptIfRunning) {
    // 只能取消还没被执行的task，即state为NEW表示还没执行
    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 {
        // 唤醒调用了get方法等待获取该任务的结果的线程，避免该任务被取消了，这些线程还在等待。
        finishCompletion();
    }
    return true;
}

  • 在Executor的工作线程执行这个任务时，会调用FutureTask的run方法，在run方法中是先检查任务的状态state，如果发现不是NEW，即可能是CANCELLED，INTERRUPTING等，则直接返回，退出该任务的执行。

    public void run() {
    // 赋值runner，指向线程池中当前执行这个task的线程
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    }
    
    ...
    
}
本文标题：JDK1.8源码分析：Future和FutureTask-任务异步执行结果
本文链接：https://blog.quwenai.cn/post/3276.html
作者授权：除特别说明外，本文由 toobug 原创编译并授权 TOOBUG信息安全网 刊载发布。
版权声明：本文不使用任何协议授权，您可以任何形式自由转载或使用。

本文链接：https://blog.quwenai.cn/post/3276.html 

分享到：
本站所有内容均来自于互联网收集及用户投稿,如果想系统学习相关知识 建议在本博客【自学成才】栏目选择合适的课程进行系统化学习
-- 展开阅读全文 --