揭密FutureTask

時間  2019-12-10
標籤 futuretask 简体版
原文   原文鏈接

      Future是咱們在使用java實現異步時最經常使用到的一個類，咱們能夠向線程池提交一個Callable，並經過future對象獲取執行結果。本篇文章主要講述了JUC中FutureTask中的一些實現原理。使用的jdk版本是1.7。java

Future

　　Future是一個接口，它定義了5個方法：異步

boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled(); 
boolean isDone(); 
V get() throws InterruptedException, ExecutionException; 
V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;　

    簡單說明一下接口定義ide

  • boolean cancel(boolean mayInterruptInRunning) 取消一個任務，並返回取消結果。參數表示是否中斷線程。
  • boolean isCancelled()                                        判斷任務是否被取消
  • Boolean isDone() 　　                                       判斷當前任務是否執行完畢，包括正常執行完畢、執行異常或者任務取消。
  • V get()                                                            獲取任務執行結果，任務結束以前會阻塞。
  • V get(long timeout, TimeUnit unit)                    在指定時間內嘗試獲取執行結果。若超時則拋出超時異常 

    寫個簡單demo：測試

public class FutureDemo {
      public static void main(String[] args) {
          ExecutorService executorService = Executors.newCachedThreadPool();
          Future future = executorService.submit(new Callable<Object>() {
              @Override
              public Object call() throws Exception {
                  Long start = System.currentTimeMillis();
                  while (true) {
                      Long current = System.currentTimeMillis();
                     if ((current - start) > 1000) {
                         return 1;
                     }
                 }
             }
         });
 
         try {
             Integer result = (Integer)future.get();
             System.out.println(result);
         }catch (Exception e){
             e.printStackTrace();
         }
     }
}　

　這裏模擬了1s鐘的CPU空轉，當執行future.get()的時候，主線程阻塞了大約一秒後得到結果。this

   固然咱們也可使用get(long timeout, TimeUnit unit)spa

try {
        Integer result = (Integer) future.get(500, TimeUnit.MILLISECONDS);                  
        System.out.println(result);
    } catch (Exception e) {
        e.printStackTrace();
    }　

     因爲在500ms內沒有結果返回，因此拋出異常，打印異常堆棧以下線程

     

      固然，若是咱們把超時時間設置的長一些，仍是能夠獲得預期的結果的。翻譯

FutureTask實現原理

    下面咱們介紹一下FutureTask內部的一些實現機制。下文從如下幾點敘述：3d

  1. 類繼承結構
  2. 核心成員變量
  3. 內部狀態轉換
  4. 核心方法解析

1 類繼承結構

      首先咱們看一下FutureTask的繼承結構：rest

　　　

      FutureTask實現了RunnableFuture接口，而RunnableFuture繼承了Runnable和Future，也就是說FutureTask既是Runnable，也是Future。

2 核心成員變量

    FutureTask內部定義瞭如下變量，以及它們的含義以下

  • volatile int state:表示對象狀態，volatile關鍵字保證了內存可見性。futureTask中定義了7種狀態，表明了7種不一樣的執行狀態
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; //任務線程已中斷
  • Callable<V> callable:被提交的任務
  • Object outcome:任務執行結果或者任務異常
  • volatile Thread runner:執行任務的線程
  • volatile WaitNode waiters：等待節點，關聯等待線程
  • long stateOffset:state字段的內存偏移量
  • long runnerOffset：runner字段的內存偏移量
  • long waitersOffset：waiters字段的內存偏移量

    後三個字段是配合Unsafe類作CAS操做使用的。

3 內部狀態轉換

    FutureTask中使用state表示任務狀態，state值變動的由CAS操做保證原子性。

    FutureTask對象初始化時，在構造器中把state置爲爲NEW，以後狀態的變動依據具體執行狀況來定。

　  例如任務執行正常結束前，state會被設置成COMPLETING，表明任務即將完成，接下來很快就會被設置爲NARMAL或者EXCEPTIONAL，這取決於調用Runnable中的call()方法是否拋出了異常。有異常則後者，反以前者。

　　任務提交後、任務結束前取消任務，那麼有可能變爲CANCELLED或者INTERRUPTED。在調用cancel方法時，若是傳入false表示不中斷線程，state會被置爲CANCELLED，反之state先被變爲INTERRUPTING，後變爲INTERRUPTED。

     總結下，FutureTask的狀態流轉過程，能夠出現如下四種狀況：

        1. 任務正常執行並返回。 NEW -> COMPLETING -> NORMAL

　　  2. 執行中出現異常。NEW -> COMPLETING -> EXCEPTIONAL

        3. 任務執行過程當中被取消，而且不響應中斷。NEW -> CANCELLED

　　  4. 任務執行過程當中被取消，而且響應中斷。 NEW -> INTERRUPTING -> INTERRUPTED　　

4 核心方法解析

　　接下來咱們一塊兒扒一扒FutureTask的源碼。咱們先看一下任務線程是怎麼執行的。當任務被提交到線程池後，會執行futureTask的run()方法。

1 public void run()

public void run() {
        // 校驗任務狀態
        if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
　　　　　　　// double check
            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
            runner = null;
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

　　翻譯一下，這個方法經歷瞭如下幾步

  1. 校驗當前任務狀態是否爲NEW以及runner是否已賦值。這一步是防止任務被取消。
  2. double-check任務狀態state
  3. 執行業務邏輯，也就是c.call()方法被執行
  4. 若是業務邏輯異常，則調用setException方法將異常對象賦給outcome，而且更新state值
  5. 若是業務正常，則調用set方法將執行結果賦給outcome，而且更新state值

 咱們繼續往下看，setException(Throwable t)和set(V v) 具體是怎麼作的

protected void set(V v) {
        // state狀態 NEW->COMPLETING
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            // COMPLETING -> NORMAL 到達穩定狀態
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
            // 一些結束工做
            finishCompletion();
        }
    }

protected void setException(Throwable t) {
    // state狀態 NEW->COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        // COMPLETING -> EXCEPTIONAL 到達穩定狀態
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
        // 一些結束工做
        finishCompletion();
    }
}　

    code中的註釋已經寫的很清楚，故不翻譯了。狀態變動的原子性由unsafe對象提供的CAS操做保證。FutureTask的outcome變量存儲執行結果或者異常對象，會由主線程返回。

2  get()和get(long timeout, TimeUnit unit)

    任務由線程池提供的線程執行，那麼這時候主線程則會阻塞，直到任務線程喚醒它們。咱們經過get(long timeout, TimeUnit unit)方法看看是怎麼作的　　

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

　  get的源碼很簡潔，首先校驗參數，而後根據state狀態判斷是否超時，若是超時則異常，不超時則調用report(s)去獲取最終結果。

    當 s<= COMPLETING時，代表任務仍然在執行且沒有被取消。若是它爲true，那麼走到awaitDone方法。

    awaitDone是futureTask實現阻塞的關鍵方法，咱們重點關注一下它的實現原理。

     /**
     * 等待任務執行完畢,若是任務取消或者超時則中止
     * @param timed 爲true表示設置超時時間
     * @param nanos 超時時間
     * @return 任務完成時的狀態
     * @throws InterruptedException
     */
    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;
            if (s > COMPLETING) {
                // 任務可能已經完成或者被取消了
                if (q != null)
                    q.thread = null;
                return s;
            }
            else if (s == COMPLETING)
                // 可能任務線程被阻塞了,主線程讓出CPU
                Thread.yield();
            else if (q == null)
                // 等待線程節點爲空,則初始化新節點並關聯當前線程
                q = new WaitNode();
            else if (!queued)
                // 等待線程入隊列,成功則queued=true
                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
                // timed=false時會走到這裏,掛起當前線程
                LockSupport.park(this);
        }
    }

註釋裏也很清楚的寫明瞭每一步的做用，咱們以設置超時時間爲例，總結一下過程

  1. 計算deadline，也就是到某個時間點後若是尚未返回結果，那麼就超時了。
  2. 進入自旋，也就是死循環。
  3. 首先判斷是否響應線程中斷。對於線程中斷的響應每每會放在線程進入阻塞以前，這裏也印證了這一點。
  4. 判斷state值，若是>COMPLETING代表任務已經取消或者已經執行完畢，就能夠直接返回了。
  5. 若是任務還在執行，則爲當前線程初始化一個等待節點WaitNode，入等待隊列。這裏和AQS的等待隊列相似，只不過Node只關聯線程，而沒有狀態。AQS裏面的等待節點是有狀態的。
  6. 計算nanos，判斷是否已經超時。若是已經超時，則移除全部等待節點，直接返回state。超時的話，state的值仍然仍是COMPLETING。
  7. 若是還未超時，就經過LockSupprot類提供的方法在指定時間內掛起當前線程，等待任務線程喚醒或者超時喚醒。

當線程被掛起以後，若是任務線程執行完畢，就會喚醒等待線程哦。這一步就是在finishCompletion裏面作的，前面已經提到這個方法。咱們再看看這個方法具體作了哪些事吧~

    /**
     * 移除並喚醒全部等待線程,執行done,置空callable
     * nulls out callable.
     */
    private void finishCompletion() {
        //遍歷等待節點
        for (WaitNode q; (q = waiters) != null;) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (;;) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        //喚醒等待線程
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    // unlink to help gc
                    q.next = null; 
                    q = next;
                }
                break;
            }
        }
        //模板方法,能夠被覆蓋
        done();
        //清空callable
        callable = null; 
    }

由代碼和註釋能夠看出來，這個方法的做用主要在於喚醒等待線程。由前文可知，當任務正常結束或者異常時，都會調用finishCompletion去喚醒等待線程。這個時候，等待線程就能夠醒來，開開心心的得到結果啦。　　

最後咱們看一下任務取消　　

3 public boolean cancel(boolean mayInterruptIfRunning)

  注意，取消操做不必定會起做用，這裏咱們先貼個demo

 1 public class FutureDemo {
 2     public static void main(String[] args) {
 3         ThreadPoolExecutor executorService = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
 4         // 預建立線程
 5         executorService.prestartCoreThread();
 6 
 7         Future future = executorService.submit(new Callable<Object>() {
 8             @Override
 9             public Object call() {
10                 System.out.println("start to run callable");
11                 Long start = System.currentTimeMillis();
12                 while (true) {
13                     Long current = System.currentTimeMillis();
14                     if ((current - start) > 1000) {
15                         System.out.println("當前任務執行已經超過1s");
16                         return 1;
17                     }
18                 }
19             }
20         });
21 
22         System.out.println(future.cancel(false));
23 
24         try {
25             Thread.currentThread().sleep(3000);
26             executorService.shutdown();
27         } catch (Exception e) {
28             //NO OP
29         }
30     }
31 }

咱們屢次測試後發現，出現了2種打印結果，如圖

　　　　　　　　　　　　　　　　結果1

　　　　　　　　　　　　　　　　結果2　　　　

第一種是任務壓根沒取消，第二種則是任務壓根沒提交成功。　

    方法簽名註釋告訴咱們，取消操做是可能會失敗的，若是當前任務已經結束或者已經取消，則當前取消操做會失敗。若是任務還沒有開始，那麼任務不會被執行。這就解釋了出現上圖結果2的狀況。咱們仍是從源碼去分析cancel()究竟作了哪些事。

public boolean cancel(boolean mayInterruptIfRunning) {
        if (state != NEW)
            return false;
        if (mayInterruptIfRunning) {
            if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
                return false;
            Thread t = runner;
            if (t != null)
                t.interrupt();
            UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
        }
        else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
            return false;
        finishCompletion();
        return true;
    }

　　執行邏輯以下

  1. state不爲NEW時，任務即將進入終態，直接返回false代表取消操做失敗。
  2. state狀態爲NEW，任務可能已經開始執行，也可能還未開始。
  3. mayInterruptIfRunning代表是否中斷線程。如果，則嘗試將state設置爲INTERRUPTING，而且中斷線程，以後將state設置爲終態INTERRUPTED。
  4. 若是mayInterruptIfRunning=false，則不中斷線程，把state設置爲CANCELLED
  5. 移除等待線程並喚醒。
  6. 返回true

    可見，cancel()方法改變了futureTask的狀態位，若是傳入的是false而且業務邏輯已經開始執行，當前任務是不會被終止的，而是會繼續執行，直到異常或者執行完畢。若是傳入的是true，會調用當前線程的interrupt()方法，把中斷標誌位設爲true。

    事實上，除非線程本身中止本身的任務，或者退出JVM，是沒有其餘方法徹底終止一個線程的任務的。mayInterruptIfRunning=true，經過但願當前線程能夠響應中斷的方式來結束任務。當任務被取消後，會被封裝爲CancellationException拋出。

總結

　　總結一下，futureTask中的任務狀態由變量state表示，任務狀態都基於state判斷。而futureTask的阻塞則是經過自旋+掛起線程實現。理解FutureTask的內部實現機制，咱們使用Future時才能更加駕輕就熟。文中摻雜着筆者的我的理解，若是有不正之處，還望讀者多多指正

 

 

做者：mayday芋頭

本博客中未標明轉載的文章歸做者mayday芋頭和博客園共有，歡迎轉載，但未經做者贊成必須保留此段聲明，且在文章頁面明顯位置給出原文鏈接，不然保留追究法律責任的權利
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程