FutureTask實現異步的分析

1.FutureTask的類結構

首先咱們看下FutureTask的類型的繼承關係,它同時實現了Future和Runnable的接口,也就是 具有了Future的異步的功能.

2. Future的定義

public interface Future<V> {

    /**
     * Attempts to cancel execution of this task.  This attempt will
     * fail if the task has already completed, has already been cancelled,
     * or could not be cancelled for some other reason. If successful,
     * and this task has not started when {@code cancel} is called,
     * this task should never run.  If the task has already started,
     * then the {@code mayInterruptIfRunning} parameter determines
     * whether the thread executing this task should be interrupted in
     * an attempt to stop the task.
     *
     * <p>After this method returns, subsequent calls to {@link #isDone} will
     * always return {@code true}.  Subsequent calls to {@link #isCancelled}
     * will always return {@code true} if this method returned {@code true}.
     *
     * @param mayInterruptIfRunning {@code true} if the thread executing this
     * task should be interrupted; otherwise, in-progress tasks are allowed
     * to complete
     * @return {@code false} if the task could not be cancelled,
     * typically because it has already completed normally;
     * {@code true} otherwise
     */
    boolean cancel(boolean mayInterruptIfRunning);

    /**
     * Returns {@code true} if this task was cancelled before it completed
     * normally.
     *
     * @return {@code true} if this task was cancelled before it completed
     */
    boolean isCancelled();

    /**
     * Returns {@code true} if this task completed.
     *
     * Completion may be due to normal termination, an exception, or
     * cancellation -- in all of these cases, this method will return
     * {@code true}.
     *
     * @return {@code true} if this task completed
     */
    boolean isDone();

    /**
     * Waits if necessary for the computation to complete, and then
     * retrieves its result.
     *
     * @return the computed result
     * @throws CancellationException if the computation was cancelled
     * @throws ExecutionException if the computation threw an
     * exception
     * @throws InterruptedException if the current thread was interrupted
     * while waiting
     */
    V get() throws InterruptedException, ExecutionException;

    /**
     * Waits if necessary for at most the given time for the computation
     * to complete, and then retrieves its result, if available.
     *
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @return the computed result
     * @throws CancellationException if the computation was cancelled
     * @throws ExecutionException if the computation threw an
     * exception
     * @throws InterruptedException if the current thread was interrupted
     * while waiting
     * @throws TimeoutException if the wait timed out
     */
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}
複製代碼

能夠看到Future是JDK1.5引入的異步的方式,而FutureTask則是異步的實現,它 提供了取消任務、檢測是取消、是否完成、獲取執行結構等操做, 其中獲取結果若是 任務沒有執行完成,會阻塞任務的執行.直到任務返回結果.

3. 任務狀態轉換

全部可能的狀態轉換:
NEW(新建立) -> COMPLETING(完成中) -> NORMAL(正常結束)
NEW(新建立) -> COMPLETING(完成中) -> EXCEPTIONAL(異常結束)
NEW(新建立) -> CANCELLED(取消)
NEW(新建立) -> INTERRUPTING(中斷中) -> INTERRUPTED(中斷完成) \

4 FutureTask任務執行

public void run() {
       if (state != NEW ||
           !RUNNER.compareAndSet(this, 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);
       }
   }
複製代碼

其實現異步的關鍵,在於最後任務執行完成, 將任務執行的執行的結果從新甚至回FutureTask的outcome字段,而後經過get方法能夠獲取異步任務返回的結果,

5 FutureTask的get方法獲取異步任務結構

public V get() throws InterruptedException, ExecutionException {
      int s = state;
      if (s <= COMPLETING)
          s = awaitDone(false, 0L);
      return report(s);
  }
複製代碼

能夠看出,在調用get方法時, 若是任務state是小於COMPLETING,說明任務還在進行中, 則調用awaitDone進行阻塞等待任務執行,

private int awaitDone(boolean timed, long nanos)
       throws InterruptedException {
       // The code below is very delicate, to achieve these goals:
       // - call nanoTime exactly once for each call to park
       // - if nanos <= 0L, return promptly without allocation or nanoTime
       // - if nanos == Long.MIN_VALUE, don't underflow
       // - if nanos == Long.MAX_VALUE, and nanoTime is non-monotonic
       //   and we suffer a spurious wakeup, we will do no worse than
       //   to park-spin for a while
       long startTime = 0L;    // Special value 0L means not yet parked
       WaitNode q = null;
       boolean queued = false;
       for (;;) {
           int s = state;
           if (s > COMPLETING) {
               if (q != null)
                   q.thread = null;
               return s;
           }
           else if (s == COMPLETING)
               // We may have already promised (via isDone) that we are done
               // so never return empty-handed or throw InterruptedException
               Thread.yield();
           else if (Thread.interrupted()) {
               removeWaiter(q);
               throw new InterruptedException();
           }
           else if (q == null) {
               if (timed && nanos <= 0L)
                   return s;
               q = new WaitNode();
           }
           else if (!queued)
               queued = WAITERS.weakCompareAndSet(this, q.next = waiters, q);
           else if (timed) {
               final long parkNanos;
               if (startTime == 0L) { // first time
                   startTime = System.nanoTime();
                   if (startTime == 0L)
                       startTime = 1L;
                   parkNanos = nanos;
               } else {
                   long elapsed = System.nanoTime() - startTime;
                   if (elapsed >= nanos) {
                       removeWaiter(q);
                       return state;
                   }
                   parkNanos = nanos - elapsed;
               }
               // nanoTime may be slow; recheck before parking
               if (state < COMPLETING)
                   LockSupport.parkNanos(this, parkNanos);
           }
           else
               LockSupport.park(this);
       }
   }
複製代碼

能夠看出這裏是經過是LockSupport.park進行現成阻塞,直到任務完成,

總結:
今天主要是分析JDK的Future實現類的FutureTask實現異步的方式,若是是向線程池提交任務,任務是有線程池的工做線程執行,而不會阻塞住線程, 可是JDK並非真正意義上的異步,由於提交任務後,當即執行get獲取結果,依然是回同步阻塞的,而Netty的Promise 則是異步的回調的方式實現異步,是JDK的異步的加強,有興趣的同窗能夠看下它的實現,之後有機會會再次分享出來.