使用双异步后 如何保证数据一致性？(2)

使用双异步后

1. 如果state小于等于COMPLETING，表示任务还在执行中；
   • 计算超时时间；
   • 如果超时，则从等待队列中移除等待节点WaitNode，返回当前状态state；
   • 阻塞队列nanos毫秒。
   • 如果已有等待节点WaitNode，将线程置空；
   • 返回当前状态；
   • 如果线程被中断，从等待队列中移除等待节点WaitNode，抛出中断异常；
   • 如果state大于COMPLETING；
   • 如果任务正在执行，让出时间片；
   • 如果还未构造等待节点，则new一个新的等待节点；
   • 如果未入队列，CAS尝试入队；
   • 如果有超时时间参数；
   • 否则阻塞队列；
2. 如果state大于COMPLETING；
   • 如果执行完毕，返回结果；
   • 如果大于等于取消状态，则抛出异常。

很多小朋友对读源码，嗤之以鼻，工作3年、5年，还是没认真读过任何源码，觉得读了也没啥用，或者读了也看不懂~

其实，只要把源码的执行流程通过画图的形式呈现出来，你就会幡然醒悟，原来是这样的~

简而言之：

1. 如果异步线程还没执行完，则进入CAS自旋；

2. 其它线程获取结果或取消后，重新唤醒CAS队列中等待的线程；

3. 再通过get()判断状态state；

4. 直至返回结果或（取消、超时、异常）为止。

二、FutureTask源码具体分析

1、FutureTask源码

通过定义整形状态值，判断state大小，这个思想很有意思，值得学习。

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

public class FutureTask<V> implements RunnableFuture<V> {
 // 最初始的状态是new 新建状态
 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; // 已中断
 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 (;;) {
         // 线程被中断，从等待队列中移除等待节点WaitNode，抛出中断异常
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }
            int s = state;
            // 任务已执行完毕或取消
            if (s > COMPLETING) {
             // 如果已有等待节点WaitNode，将线程置空
                if (q != null)
                    q.thread = null;
                return s;
            }
            // 任务正在执行，让出时间片
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
            // 还未构造等待节点，则new一个新的等待节点
            else if (q == null)
                q = new WaitNode();
            // 未入队列，CAS尝试入队
            else if (!queued)
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            // 如果有超时时间参数
            else if (timed) {
             // 计算超时时间
                nanos = deadline - System.nanoTime();
                // 如果超时，则从等待队列中移除等待节点WaitNode，返回当前状态state
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                // 阻塞队列nanos毫秒
                LockSupport.parkNanos(this, nanos);
            }
            else
             // 阻塞队列
                LockSupport.park(this);
        }
    }
    
 private V report(int s) throws ExecutionException {
  // 获取outcome中记录的返回结果
        Object x = outcome;
        // 如果执行完毕，返回结果
        if (s == NORMAL)
            return (V)x;
            // 如果大于等于取消状态，则抛出异常
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }
}