1. public class FutureTask<V> implements RunnableFuture<V> {
    2.     //表示当前task状态
    3.     private volatile int state;
    4.     //当前任务尚未执行
    5.     private static final int NEW          = 0;
    6.     //当前任务正在结束，稍微完全结束，一种临界状态
    7.     private static final int COMPLETING   = 1;
    8.     //当前任务正常结束
    9.     private static final int NORMAL       = 2;
    10.     //当前任务执行过程中发生了异常。 内部封装的 callable.run() 向上抛出异常了
    11.     private static final int EXCEPTIONAL  = 3;
    12.     //当前任务被取消
    13.     private static final int CANCELLED    = 4;
    14.     //当前任务中断中..
    15.     private static final int INTERRUPTING = 5;
    16.     //当前任务已中断
    17.     private static final int INTERRUPTED  = 6;
    19.     //submit(runnable/callable)   runnable 使用 装饰者模式 伪装成 Callable了。
    20.     private Callable<V> callable;
    22.     //正常情况下：任务正常执行结束，outcome保存执行结果。 callable 返回值。
    23.     //非正常情况：callable向上抛出异常，outcome保存异常
    24.     private Object outcome; // non-volatile, protected by state reads/writes
    26.     //当前任务被线程执行期间，保存当前执行任务的线程对象引用。
    27.     private volatile Thread runner;
    29.     //因为会有很多线程去get当前任务的结果，所以 这里使用了一种数据结构 stack 头插 头取 的一个队列。
    30.     private volatile WaitNode waiters;
    33.     @SuppressWarnings("unchecked")
    34.     private V report(int s) throws ExecutionException {
    35.         //正常情况下，outcome 保存的是callable运行结束的结果
    36.         //非正常，保存的是 callable 抛出的异常。
    37.         Object x = outcome;
    38.         //条件成立：当前任务状态正常结束
    39.         if (s == NORMAL)
    40.             //直接返回callable运算结果
    41.             return (V)x;
    43.         //被取消状态
    44.         if (s >= CANCELLED)
    45.             throw new CancellationException();
    47.         //执行到这，说明callable接口实现中，是有bug的...
    48.         throw new ExecutionException((Throwable)x);
    49.     }
    51.     /**
    52.      * Creates a {@code FutureTask} that will, upon running, execute the
    53.      * given {@code Callable}.
    54.      *
    55.      * @param  callable the callable task
    56.      * @throws NullPointerException if the callable is null
    57.      */
    58.     public FutureTask(Callable<V> callable) {
    59.         if (callable == null)
    60.             throw new NullPointerException();
    61.         //callable就是程序员自己实现的业务类
    62.         this.callable = callable;
    63.         //设置当前任务状态为 NEW
    64.         this.state = NEW;       // ensure visibility of callable
    65.     }
    67.     public FutureTask(Runnable runnable, V result) {
    68.         //使用装饰者模式将runnable转换为了 callable接口，外部线程 通过get获取
    69.         //当前任务执行结果时，结果可能为 null 也可能为 传进来的值。
    70.         this.callable = Executors.callable(runnable, result);
    71.         this.state = NEW;       // ensure visibility of callable
    72.     }
    74.     public boolean isCancelled() {
    75.         return state >= CANCELLED;
    76.     }
    78.     public boolean isDone() {
    79.         return state != NEW;
    80.     }
    82.     public boolean cancel(boolean mayInterruptIfRunning) {
    83.         //条件一：state == NEW 成立 表示当前任务处于运行中 或者 处于线程池 任务队列中..
    84.         //条件二：UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))
    85.         //      条件成立：说明修改状态成功，可以去执行下面逻辑了，否则 返回false 表示cancel失败。
    86.         if (!(state == NEW &&
    87.                 UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
    88.                         mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
    89.             return false;
    91.         try {    // in case call to interrupt throws exception
    92.             if (mayInterruptIfRunning) {
    93.                 try {
    94.                     //执行当前FutureTask 的线程，有可能现在是null，是null 的情况是： 当前任务在 队列中，还没有线程获取到它呢。。
    95.                     Thread t = runner;
    96.                     //条件成立：说明当前线程 runner ，正在执行task.
    97.                     if (t != null)
    98.                         //给runner线程一个中断信号.. 如果你的程序是响应中断 会走中断逻辑..假设你程序不是响应中断的..啥也不会发生。
    99.                         t.interrupt();
    100.                 } finally { // final state
    101.                     //设置任务状态为 中断完成。
    102.                     UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
    103.                 }
    104.             }
    106.         } finally {
    107.             //唤醒所有get() 阻塞的线程。
    108.             finishCompletion();
    109.         }
    111.         return true;
    112.     }
    114.     /**
    115.      * @throws CancellationException {@inheritDoc}
    116.      */
    117.     //场景：多个线程等待当前任务执行完成后的结果...
    118.     public V get() throws InterruptedException, ExecutionException {
    119.         //获取当前任务状态
    120.         int s = state;
    121.         //条件成立：未执行、正在执行、正完成。 调用get的外部线程会被阻塞在get方法上。
    122.         if (s <= COMPLETING)
    123.             //返回task当前状态，可能当前线程在里面已经睡了一会了..
    124.             s = awaitDone(false, 0L);
    126.         return report(s);
    127.     }
    129.     /**
    130.      * @throws CancellationException {@inheritDoc}
    131.      */
    132.     public V get(long timeout, TimeUnit unit)
    133.             throws InterruptedException, ExecutionException, TimeoutException {
    134.         if (unit == null)
    135.             throw new NullPointerException();
    136.         int s = state;
    137.         if (s <= COMPLETING &&
    138.                 (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
    139.             throw new TimeoutException();
    140.         return report(s);
    141.     }
    144.     protected void done() { }
    146.     protected void set(V v) {
    147.         //使用CAS方式设置当前任务状态为 完成中..
    148.         //有没有可能失败呢？ 外部线程等不及了，直接在set执行CAS之前 将  task取消了。  很小概率事件。
    149.         if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    151.             outcome = v;
    152.             //将结果赋值给 outcome之后，马上会将当前任务状态修改为 NORMAL 正常结束状态。
    153.             UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
    155.             //猜一猜？
    156.             //最起码得把get() 再此阻塞的线程 唤醒..
    157.             finishCompletion();
    158.         }
    159.     }
    161.     protected void setException(Throwable t) {
    162.         //使用CAS方式设置当前任务状态为 完成中..
    163.         //有没有可能失败呢？ 外部线程等不及了，直接在set执行CAS之前 将  task取消了。  很小概率事件。
    164.         if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    165.             //引用的是 callable 向上层抛出来的异常。
    166.             outcome = t;
    167.             //将当前任务的状态 修改为 EXCEPTIONAL
    168.             UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
    169.             //回头讲完 get() 。
    170.             finishCompletion();
    171.         }
    172.     }
    174.     //submit(runnable/callable) -> newTaskFor(runnable) -> execute(task)   -> pool
    175.     //任务执行入口
    176.     public void run() {
    177.         //条件一：state != NEW 条件成立，说明当前task已经被执行过了 或者 被cancel 了，总之非NEW状态的任务，线程就不处理了。
    178.         //条件二：!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread())
    179.         //       条件成立：cas失败，当前任务被其它线程抢占了...
    180.         if (state != NEW ||
    181.                 !UNSAFE.compareAndSwapObject(this, runnerOffset,
    182.                         null, Thread.currentThread()))
    183.             return;
    185.         //执行到这里，当前task一定是 NEW 状态，而且 当前线程也抢占TASK成功！
    187.         try {
    188.             //callable 就是程序员自己封装逻辑的callable 或者 装饰后的runnable
    189.             Callable<V> c = callable;
    190.             //条件一：c != null 防止空指针异常
    191.             //条件二：state == NEW 防止外部线程 cancel掉当前任务。
    192.             if (c != null && state == NEW) {
    194.                 //结果引用
    195.                 V result;
    196.                 //true 表示callable.run 代码块执行成功 未抛出异常
    197.                 //false 表示callable.run 代码块执行失败 抛出异常
    198.                 boolean ran;
    200.                 try {
    201.                     //调用程序员自己实现的callable 或者 装饰后的runnable
    202.                     result = c.call();
    203.                     //c.call未抛出任何异常，ran会设置为true 代码块执行成功
    204.                     ran = true;
    205.                 } catch (Throwable ex) {
    206.                     //说明程序员自己写的逻辑块有bug了。
    207.                     result = null;
    208.                     ran = false;
    209.                     setException(ex);
    210.                 }
    214.                 if (ran)
    215.                     //说明当前c.call正常执行结束了。
    216.                     //set就是设置结果到outcome
    217.                     set(result);
    218.             }
    219.         } finally {
    220.             // runner must be non-null until state is settled to
    221.             // prevent concurrent calls to run()
    222.             runner = null;
    223.             // state must be re-read after nulling runner to prevent
    224.             // leaked interrupts
    225.             int s = state;
    226.             if (s >= INTERRUPTING)
    227.                 //回头再说..讲了 cancel() 就明白了。
    228.                 handlePossibleCancellationInterrupt(s);
    229.         }
    230.     }
    232.     protected boolean runAndReset() {
    233.         if (state != NEW ||
    234.                 !UNSAFE.compareAndSwapObject(this, runnerOffset,
    235.                         null, Thread.currentThread()))
    236.             return false;
    237.         boolean ran = false;
    238.         int s = state;
    239.         try {
    240.             Callable<V> c = callable;
    241.             if (c != null && s == NEW) {
    242.                 try {
    243.                     c.call(); // don't set result
    244.                     ran = true;
    245.                 } catch (Throwable ex) {
    246.                     setException(ex);
    247.                 }
    248.             }
    249.         } finally {
    250.             // runner must be non-null until state is settled to
    251.             // prevent concurrent calls to run()
    252.             runner = null;
    253.             // state must be re-read after nulling runner to prevent
    254.             // leaked interrupts
    255.             s = state;
    256.             if (s >= INTERRUPTING)
    257.                 handlePossibleCancellationInterrupt(s);
    258.         }
    259.         return ran && s == NEW;
    260.     }
    262.     /**
    263.      * Ensures that any interrupt from a possible cancel(true) is only
    264.      * delivered to a task while in run or runAndReset.
    265.      */
    266.     private void handlePossibleCancellationInterrupt(int s) {
    267.         // It is possible for our interrupter to stall before getting a
    268.         // chance to interrupt us.  Let's spin-wait patiently.
    269.         if (s == INTERRUPTING)
    270.             while (state == INTERRUPTING)
    271.                 Thread.yield(); // wait out pending interrupt
    273.         // assert state == INTERRUPTED;
    275.         // We want to clear any interrupt we may have received from
    276.         // cancel(true).  However, it is permissible to use interrupts
    277.         // as an independent mechanism for a task to communicate with
    278.         // its caller, and there is no way to clear only the
    279.         // cancellation interrupt.
    280.         //
    281.         // Thread.interrupted();
    282.     }
    284.     /**
    285.      * Simple linked list nodes to record waiting threads in a Treiber
    286.      * stack.  See other classes such as Phaser and SynchronousQueue
    287.      * for more detailed explanation.
    288.      */
    289.     static final class WaitNode {
    290.         volatile Thread thread;
    291.         volatile WaitNode next;
    292.         WaitNode() { thread = Thread.currentThread(); }
    293.     }
    295.     /**
    296.      * Removes and signals all waiting threads, invokes done(), and
    297.      * nulls out callable.
    298.      */
    299.     private void finishCompletion() {
    300.         // assert state > COMPLETING;
    301.         //q指向waiters 链表的头结点。
    302.         for (WaitNode q; (q = waiters) != null;) {
    304.             //使用cas设置 waiters 为 null 是因为怕 外部线程使用 cancel 取消当前任务 也会触发finishCompletion方法。 小概率事件。
    305.             if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
    306.                 for (;;) {
    307.                     //获取当前node节点封装的 thread
    308.                     Thread t = q.thread;
    309.                     //条件成立：说明当前线程不为null
    310.                     if (t != null) {
    312.                         q.thread = null;//help GC
    313.                         //唤醒当前节点对应的线程
    314.                         LockSupport.unpark(t);
    315.                     }
    316.                     //next 当前节点的下一个节点
    317.                     WaitNode next = q.next;
    319.                     if (next == null)
    320.                         break;
    322.                     q.next = null; // unlink to help gc
    323.                     q = next;
    324.                 }
    326.                 break;
    327.             }
    328.         }
    330.         done();
    332.         //将callable 设置为null helpGC
    333.         callable = null;        // to reduce footprint
    334.     }
    336.     private int awaitDone(boolean timed, long nanos)
    337.             throws InterruptedException {
    338.         //0 不带超时
    339.         final long deadline = timed ? System.nanoTime() + nanos : 0L;
    340.         //引用当前线程 封装成 WaitNode 对象
    341.         WaitNode q = null;
    342.         //表示当前线程 waitNode对象 有没有 入队/压栈
    343.         boolean queued = false;
    344.         //自旋
    345.         for (;;) {
    347.             //条件成立：说明当前线程唤醒 是被其它线程使用中断这种方式喊醒的。interrupted()
    348.             //返回true 后会将 Thread的中断标记重置回false.
    349.             if (Thread.interrupted()) {
    350.                 //当前线程node出队
    351.                 removeWaiter(q);
    352.                 //get方法抛出 中断异常。
    353.                 throw new InterruptedException();
    354.             }
    356.             //假设当前线程是被其它线程 使用unpark(thread) 唤醒的话。会正常自旋，走下面逻辑。
    358.             //获取当前任务最新状态
    359.             int s = state;
    360.             //条件成立：说明当前任务 已经有结果了.. 可能是好 可能是 坏..
    361.             if (s > COMPLETING) {
    363.                 //条件成立：说明已经为当前线程创建过node了，此时需要将 node.thread = null helpGC
    364.                 if (q != null)
    365.                     q.thread = null;
    366.                 //直接返回当前状态.
    367.                 return s;
    368.             }
    369.             //条件成立：说明当前任务接近完成状态...这里让当前线程再释放cpu ，进行下一次抢占cpu。
    370.             else if (s == COMPLETING) // cannot time out yet
    371.                 Thread.yield();
    372.             //条件成立：第一次自旋，当前线程还未创建 WaitNode 对象，此时为当前线程创建 WaitNode对象
    373.             else if (q == null)
    374.                 q = new WaitNode();
    375.             //条件成立：第二次自旋，当前线程已经创建 WaitNode对象了，但是node对象还未入队
    376.             else if (!queued){
    377.                 //当前线程node节点 next 指向 原 队列的头节点   waiters 一直指向队列的头！
    378.                 q.next = waiters;
    379.                 //cas方式设置waiters引用指向 当前线程node， 成功的话 queued == true 否则，可能其它线程先你一步入队了。
    380.                 queued = UNSAFE.compareAndSwapObject(this, waitersOffset, waiters, q);
    381.             }
    382.             //第三次自旋，会到这里。
    383.             else if (timed) {
    384.                 nanos = deadline - System.nanoTime();
    385.                 if (nanos <= 0L) {
    386.                     removeWaiter(q);
    387.                     return state;
    388.                 }
    389.                 LockSupport.parkNanos(this, nanos);
    390.             }
    391.             else
    392.                 //当前get操作的线程就会被park了。  线程状态会变为 WAITING状态，相当于休眠了..
    393.                 //除非有其它线程将你唤醒  或者 将当前线程 中断。
    394.                 LockSupport.park(this);
    395.         }
    396.     }
    398.     private void removeWaiter(WaitNode node) {
    399.         if (node != null) {
    400.             node.thread = null;
    401.             retry:
    402.             for (;;) {          // restart on removeWaiter race
    403.                 for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
    404.                     s = q.next;
    405.                     if (q.thread != null)
    406.                         pred = q;
    407.                     else if (pred != null) {
    408.                         pred.next = s;
    409.                         if (pred.thread == null) // check for race
    410.                             continue retry;
    411.                     }
    412.                     else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
    413.                             q, s))
    414.                         continue retry;
    415.                 }
    416.                 break;
    417.             }
    418.         }
    419.     }
    421.     // Unsafe mechanics
    422.     private static final sun.misc.Unsafe UNSAFE;
    423.     private static final long stateOffset;
    424.     private static final long runnerOffset;
    425.     private static final long waitersOffset;
    426.     static {
    427.         try {
    428.             UNSAFE = sun.misc.Unsafe.getUnsafe();
    429.             Class<?> k = FutureTask.class;
    430.             stateOffset = UNSAFE.objectFieldOffset
    431.                     (k.getDeclaredField("state"));
    432.             runnerOffset = UNSAFE.objectFieldOffset
    433.                     (k.getDeclaredField("runner"));
    434.             waitersOffset = UNSAFE.objectFieldOffset
    435.                     (k.getDeclaredField("waiters"));
    436.         } catch (Exception e) {
    437.             throw new Error(e);
    438.         }
    439.     }
    441. }