JUC-21-FutureTask 源码分析

JUC-21-FutureTask 源码分析

• 废话不多收，直接手撕源码

package com.zhuuu.sources;

import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;

/**
 * @since 1.5
 * @author Doug Lea
 * @param <V> The result type returned by this FutureTask's {@code get} methods
 */
public class FutureTask<V> implements RunnableFuture<V> {
    /**
     * Possible state transitions:
     * NEW -> COMPLETING -> NORMAL
     * NEW -> COMPLETING -> EXCEPTIONAL
     * NEW -> CANCELLED
     * NEW -> INTERRUPTING -> INTERRUPTED
     */
    private volatile int state;                     // 判断当前task的状态
    private static final int NEW          = 0;      // 当前任务尚未执行
    private static final int COMPLETING   = 1;      // 当前任务正在结束，稍微完全结束的临界状态
    private static final int NORMAL       = 2;      // 当前任务正常结束
    private static final int EXCEPTIONAL  = 3;      // 当前任务执行过程中出现了异常，Callable.run()方法向上抛出异常
    private static final int CANCELLED    = 4;      // 当前任务被取消
    private static final int INTERRUPTING = 5;      // 当前任务中断中
    private static final int INTERRUPTED  = 6;      // 当前任务已中断

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

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




    // submit(runnable/callable) runnable 使用装饰着伪装成 callable了
    private Callable<V> callable;
    // 1. 正常情况下：任务执行结束，outcome保存执行结果，callable 返回值
    // 2. 非正常情况下，callable向上抛出异常，outcome保存异常
    private Object outcome;
    // 当前任务执行期间，保存当前执行对象线程的引用
    private volatile Thread runner;
    // 所有等待的线程，包装成一个节点，并且把自己unsafe.park掉,除非被其他线程唤醒或者中断
    // 因为会有很多线程去get当前任务的结果，所有这里使用了一个头插头取的队列
    private volatile WaitNode waiters;                                           // waiters保存了一个头节点的引用


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



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




    @SuppressWarnings("unchecked")
    // get()方法向上汇报结果
    private V report(int s) throws ExecutionException {
        // 1. 正常情况下，outcome保存的是callable()运行结束的结果
            //非正常情况下，这里保存的是callale()抛出的异常
        Object x = outcome;
        // 2. 当前任务正常结束，直接返回结果
        if (s == NORMAL)
            return (V)x;
        // 3. 非正常结束，被中断或者被取消了，抛出了CancellationException
        if (s >= CANCELLED)
            throw new CancellationException();

        // 4. 执行到这里，抛出ExecutionException，说明callable接口实现中是有bug的
        throw new ExecutionException((Throwable)x);
    }





    //  构造方法，大多数情况不会来new它,而是直接提交到线程池里面
    //  submit 最终调用的是AbstractExecuotrService 的 execute
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        // 1. callable 就是程序员自己实现的业务类
        this.callable = callable;
        // 2. 设置当前任务的状态为new
        this.state = NEW;       // ensure visibility of callable
    }
    /**
     * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
     * @throws NullPointerException if the runnable is null
     */
    public FutureTask(Runnable runnable, V result) {
        // 1. 使用适配器模式将runnable 将转换成 callable，外部线程通过get获取（结果可能是null 也可能是自己 传的值）
        // 2. 传的结果是什么，返回的结果是什么，无法拿到线程执行的结果(自己骗自己)
        this.callable = Executors.callable(runnable, result);
        // 3. 设置任务的状态是new
        this.state = NEW;       // ensure visibility of callable
    }







    // cancel()使用场景 ： 提交之后通过线程池执行任务，这个任务一直不结束，外层也不知道到里面是否执行完。（通过外部Future句柄中断掉）
    public boolean cancel(boolean mayInterruptIfRunning) {
        // 1. 条件一 ： ！（state == NEW） 表示当前任务处理运行中 或者 处理线程池的任务队列中，才是可以取消的
        //    条件二 ： UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
        //                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED))
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;                              // 否则返回false,表示cancel()失败

        // 2. 只有上面执行成功，才会执行下面的逻辑
        try {
            // 2.1 运行中打断，给当前的线程发送一个中断信号
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;                 // 拿到执行当前futureTask的线程
                    // 2.1.1 当前执行futureTask的线程为null的情况，说明当前任务在队列中呢，还没有线程有空来执行它
                    // 2.1.1 这里不是null 说明当前线程正在执行，给当前线程一个中断信号
                            // 这个时候如果程序响应中断就走中断逻辑
                            // 如果程序不响应中断，那么啥也不会发生
                    if (t != null)
                        t.interrupt();
                }
                // 2.2 设置任务状态为中断完成
                finally { // final state
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }

            // 3. 最后唤醒所有get阻塞的线程
                // 虽然对于2来说，如果程序不响应中断，中断什么的也不会发生，但是这里还是会唤醒所有阻塞的线程
        } finally {
            finishCompletion();
        }
        return true;
    }









    // 重要的get()方法 ： 这里要注意的是，不要认为只有一个线程会来这里调用get（）方法，可能会有多个线程来这里调用get()方法
    // 场景： 多个线程等待当前任务执行完成后的结果
    public V get() throws InterruptedException, ExecutionException {
        // 1.1 获取当前任务状态
        int s = state;
        // 1.2 条件成立：要么是未执行，要么是正在执行，要么是正在完成
        //     调用get()的外部线程，会在这里阻塞等待结果
        //     这里尤其要注意的是，这里get()的线程和runner对应执行任务的线程，一定不是同一个线程
        if (s <= COMPLETING)
            // 1.3 整个futureTask 最核心的方法
            // 这里返回当前task的状态，或者这里线程已经抛出中断异常了
            s = awaitDone(false, 0L);
        return report(s);
    }

    /**
     * @throws CancellationException {@inheritDoc}
     */
    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);
    }





    protected void done() { }




    // 设置c.call() 结果结果的方法
    protected void set(V v) {
        // 1.1 使用CAS方式设置当前任务状态为完成中，临界值的状态
        // 1.2 这里有没有可能失败呢？ 外部线程等不及了，直接cancel()掉task。基本不可能发生
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            // 2.1 结果赋值给outcome
            outcome = v;
            // 2.2 将结果赋值给outcome之后，马上直接改写内存中的Int值，不是CAS。将当前任务转换成Normal，正常结束的状态
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            // 2.3 回头再说，讲完get（）再说，猜一猜会做什么事
                // 起码会把get() 阻塞的线程给唤醒
            finishCompletion();
        }
    }



    // 设置c.call() 设置异常的方法
    protected void setException(Throwable t) {
        // 1.1 使用CAS方式设置当前任务状态为完成中，临界值的状态
        // 1.2 这里有没有可能失败呢？ 外部线程等不及了，直接cancel()掉task。基本不可能发生
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            // 2.1 引用的是 callable向上层抛出的异常，拿不到结果了，只能get到异常
            outcome = t;
            // 2.2 将结果赋值给outcome之后，马上直接改写内存中的Int值，不是CAS。将当前任务转换成EXCEPTIONAL
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            // 2.3 回头再说，讲完get（）再说
            finishCompletion();
        }
    }




    // 核心方法 ： 线程执行方法的入口
    // 提交任务的流程  ： sumbit(runnable/callable) -> newTaskFor(runnable) -> execute(task) ->pool
    public void run() {
        // 条件1 ： state != new : 说明当前task已经被执行过了或者被cancel掉了，总之非new状态的任务，线程就不处理了
        // 条件2 ： !UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread())
                    // 2.1 通过CAS 将当前线程设置到FutureTask的runner中
                    // 2.2 若这里条件成立，说明cas失败，也就是当前任务被其他线程给抢占了
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;


        // 前置条件：执行到这里，当前的task一定是New 状态，而且当前线程也抢占到task成功
        try {
            // 1.1 这里的callable就是程序员自己设置的任务，或者是适配后的runnable
            Callable<V> c = callable;
            // 条件1： c != null ，防止空指针异常
            // 条件2： state == NEW ，防止外部线程在这个空隙中cancel掉任务
            if (c != null && state == NEW) {
                // 2.1 保留结果的引用
                V result;

                // 2.2 true 表示 callable.run正常执行成功， false表示callable.run 执行失败
                boolean ran;

                try {
                    // 2.3 正常执行,ran 是 true
                    // 拿到callable就是程序员自己设置的任务，或者是适配后的runnable 的运行结果
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    // 2.4 执行失败，ran 默认是false
                    // 结果设置为空
                    result = null;
                    ran = false;
                    // 2.5 回头再说
                    setException(ex);
                }

                // 3. 说明当前c.call()正常执行结束了，设置结果
                if (ran)
                    // set设置结果到outcome的一个过程，点开源码
                    set(result);
            }

            // 4. finally块 ： 这里只有抛出异常才会来到这里
        } finally {
            // 4.1 将当前执行task的线程给设置为null
            runner = null;
            int s = state;

            // 4.2 大于等于中断中，稍微让当前线程释放CPU
            if (s >= INTERRUPTING)
                // 4.3 回头再说，讲完 cancel() 方法就明白了
                handlePossibleCancellationInterrupt(s);
        }
    }


    // 任务完成之后，在改完任务状态为normal之后，需要调用此方法
    /**
     * Removes and signals all waiting threads, invokes done(), and
     * nulls out callable.
     */
    private void finishCompletion() {
        // 1. q指向waiters链表的头节点，头节点不为null的情况
        for (WaitNode q; (q = waiters) != null;) {

            // 2. 通过CAS方式，首先将CAS将waiters头节点设置为null
            // 这里是用CAS是因为怕外部线程，通过cancel()取消当前任务，也会触发finishCompletion（）
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {


                for (;;) {
                    // 2.1 首先拿到当前节点的线程
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;             // 将当前节点的线程置为空（help GC）
                        LockSupport.unpark(t);       // 并且唤醒所有在awaitDone阻塞的线程,unpark的方式，哪里阻塞在哪里唤醒
                    }
                    // 2.2 处理当前节点的下一个节点
                    WaitNode next = q.next;
                    // 2.2.1 下一个节点为null的话，说明处理的是最后一个节点，就都唤醒完了
                    if (next == null)
                        break;
                    // 2.2.2 如果当前节点next不为空，继续处理下一个节点去
                    q.next = null; // unlink to help gc
                    q = next;
                }
                // 2.3 都处理完了，跳出自旋
                break;
            }
        }

        // 3. 啥也没做
        done();

        // 4. 设置当前任务为空
        callable = null;        // to reduce footprint
    }





    // 整个FutureTask 最核心的方法，这里讲的不带超时时间
    // 从FutureTask.get()方法入口进来的
    private int awaitDone(boolean timed, long nanos)  throws InterruptedException {
        // 1.1 timed = 0 不带超时时间
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        // 1.2 初始：引用当前线程，封装成WaitNode对象
        WaitNode q = null;
        // 1.3 表示当前线程waitNode对象有没有入队
        boolean queued = false;

        // 2. 进入一个自旋，源码从下往上看
        for (;;) {
            // 2.5 从LockSupport.park中断的线程
            // 条件成立：说明当前线程是被其他线程使用中断方式喊醒的。
            // Thread.interrupted()会返回true,并且把当前线程的中断标记位重新设置为false
            if (Thread.interrupted()) {
                // 当前线程出队的操作
                removeWaiter(q);
                // 向外层get()方法抛出中断异常
                throw new InterruptedException();
            }
            // 2.6 从LockSupport.park 被其他线程唤醒的线程 unpark，就会走下面的逻辑
            // 条件成立：unpark唤醒后会正常自旋，走下面的逻辑
            // 获取当前线程最新的状态
            int s = state;
            // 条件成立：说明当前任务已经有结果了，可能是好结果或者坏结果
            if (s > COMPLETING) {
                // 条件成立，说明当前线程已经创建过node了，此时需要将node中的Thread指向null, (help GC)
                if (q != null)
                    q.thread = null;
                // 直接返回当前状态
                return s;
            }
            // 2.7 说明当前任务接近完成状态，这里让当前线程再释放CPU,进行下一次抢占CPU
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();



                // 2.1 条件成立：第一次自旋，当前线程还未创建WaitNode对象，此时为当前线程创建WaitNode对象
            else if (q == null)
                q = new WaitNode();

                // 2.2 条件成立：第二次自旋，当前线程创建WaitNode对象，但是当前node还未入队
            else if (!queued) {
                // 2.2.1 当前线程node节点 next指向原队列的头节点 ,waiters一直指向队列的头
                q.next = waiters;
                // 2.2.2 CAS方式设置waiters引用指向当前线程node,成功的话queue == true,否则的话，可能其他线程先你一步入队了
                //          如果失败了的话，还会自旋去重新尝试入队的，一直成功位置
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset, waiters, q);
            }
            // 2.3 条件成立：第三次自旋，会来到这里。这里的timed是0,会走到下一个else里面
            else if (timed) {
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                LockSupport.parkNanos(this, nanos);
            }
            // 2.4 当前get操作的线程就会被park掉了，线程的状态会变成WAITING状态，相当于休眠了。。
            //     除非其他线程将你 唤醒，或者将当前线程  中断
            else
                LockSupport.park(this);
        }
    }




    // 出队的方法
    private void removeWaiter(WaitNode node) {
        if (node != null) {
            // 1. 首先将引用的线程置为空
            node.thread = null;

            // 2. 自旋,断开链表的操作
            retry:
            for (;;) {
                // 3. 内层for循环
                // pred当前节点上一个节点，q当前节点，s当前节点下一个节点。
                // 循环条件：q != null 一轮结束后q = s
                for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                    s = q.next;                    // 拿到当前节点的下一个节点

                    // 3.1 如果当前节点不为空
                    if (q.thread != null)
                        pred = q;                  // 指针前进，遍历到下一个节点去
                        // 3.2 当前节点q的线程是null，说明是一个要移除出队的节点
                        // 同时如果pred不是null的情况，说明当前节点不是头节点
                    else if (pred != null) {
                        pred.next = s;             // 前置节点的节点直接指向下一个节点，跳过该节点，相当于直接删除这个节点
                        if (pred.thread == null)   // 怕的就是pred这个线程也执行出队了，重新再来一次
                            continue retry;
                    }
                    // 3.3 如果当前节点的线程是null, 同时又是头节点
                    // CAS 当前节点直接指向下一个节点，跳过该节点，相当于直接删除这个节点
                    else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                            q, s))
                        continue retry;             // 成功就接着往下的逻辑，看有没有还要出队的
                }
                break;
            }
        }
    }






    /**
     * Executes the computation without setting its result, and then
     * resets this future to initial state, failing to do so if the
     * computation encounters an exception or is cancelled.  This is
     * designed for use with tasks that intrinsically execute more
     * than once.
     *
     * @return {@code true} if successfully run and reset
     */
    protected boolean runAndReset() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return false;
        boolean ran = false;
        int s = state;
        try {
            Callable<V> c = callable;
            if (c != null && s == NEW) {
                try {
                    c.call(); // don't set result
                    ran = true;
                } catch (Throwable ex) {
                    setException(ex);
                }
            }
        } 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
            s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
        return ran && s == NEW;
    }

    /**
     * Ensures that any interrupt from a possible cancel(true) is only
     * delivered to a task while in run or runAndReset.
     */
    private void handlePossibleCancellationInterrupt(int s) {
        // It is possible for our interrupter to stall before getting a
        // chance to interrupt us.  Let's spin-wait patiently.
        if (s == INTERRUPTING)
            while (state == INTERRUPTING)
                Thread.yield(); // wait out pending interrupt

        // assert state == INTERRUPTED;

        // We want to clear any interrupt we may have received from
        // cancel(true).  However, it is permissible to use interrupts
        // as an independent mechanism for a task to communicate with
        // its caller, and there is no way to clear only the
        // cancellation interrupt.
        //
        // Thread.interrupted();
    }


}

打赏