死磕 java線程系列之線程池深刻解析——將來任務執行流程

時間 2019-11-07

標籤 java 線程系列深刻解析將來任務執行流程欄目 Java 简体版

原文原文鏈接

（手機橫屏看源碼更方便）java

注：java源碼分析部分如無特殊說明均基於 java8 版本。設計模式

注：線程池源碼部分如無特殊說明均指ThreadPoolExecutor類。框架

簡介

前面咱們一塊兒學習了線程池中普通任務的執行流程，但其實線程池中還有一種任務，叫做將來任務（future task），使用它您能夠獲取任務執行的結果，它是怎麼實現的呢？異步

建議學習本章前先去看看彤哥以前寫的《死磕 java線程系列之本身動手寫一個線程池（續）》，有助於理解本章的內容，且那邊的代碼比較短小，學起來相對容易一些。源碼分析

問題

（1）線程池中的將來任務是怎麼執行的？學習

（2）咱們能學到哪些比較好的設計模式？this

（3）對咱們將來學習別的框架有什麼幫助？線程

來個栗子

咱們仍是從一個例子入手，來說解來章的內容。設計

咱們定義一個線程池，並使用它提交5個任務，這5個任務分別返回0、一、二、三、4，在將來的某一時刻，咱們再取用它們的返回值，作一個累加操做。code

public class ThreadPoolTest02 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 新建一個固定5個線程的線程池
        ExecutorService threadPool = Executors.newFixedThreadPool(5);

        List<Future<Integer>> futureList = new ArrayList<>();
        // 提交5個任務，分別返回0、一、二、三、4
        for (int i = 0; i < 5; i++) {
            int num = i;

            // 任務執行的結果用Future包裝
            Future<Integer> future = threadPool.submit(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("return: " + num);
                // 返回值
                return num;
            });

            // 把future添加到list中
            futureList.add(future);
        }

        // 任務所有提交完再從future中get返回值，並作累加
        int sum = 0;
        for (Future<Integer> future : futureList) {
            sum += future.get();
        }

        System.out.println("sum=" + sum);
    }
}

這裏咱們思考兩個問題：

（1）若是這裏使用普通任務，要怎麼寫，時間大概是多少？

若是使用普通任務，那麼就要把累加操做放到任務裏面，並且並非那麼好寫（final的問題），總時間大概是1秒多一點。可是，這樣有一個缺點，就是累加操做跟任務自己的內容耦合到一塊兒了，後面若是改爲累乘，還要修改任務的內容。

（2）若是這裏把future.get()放到for循環裏面，時間大概是多少？

這個問題咱們先不回答，先來看源碼分析。

submit()方法

submit方法，它是提交有返回值任務的一種方式，內部使用將來任務（FutureTask）包裝，再交給execute()去執行，最後返回將來任務自己。

public <T> Future<T> submit(Callable<T> task) {
    // 非空檢測
    if (task == null) throw new NullPointerException();
    // 包裝成FutureTask
    RunnableFuture<T> ftask = newTaskFor(task);
    // 交給execute()方法去執行
    execute(ftask);
    // 返回futureTask
    return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    // 將普通任務包裝成FutureTask
    return new FutureTask<T>(callable);
}

這裏的設計很巧妙，實際上這兩個方法都是在AbstractExecutorService這個抽象類中完成的，這是模板方法的一種運用。

咱們來看看FutureTask的繼承體系：

FutureTask實現了RunnableFuture接口，而RunnableFuture接口組合了Runnable接口和Future接口的能力，而Future接口提供了get任務返回值的能力。

問題：submit()方法返回的爲何是Future接口而不是RunnableFuture接口或者FutureTask類呢？

答：這是由於submit()返回的結果，對外部調用者只想暴露其get()的能力（Future接口），而不想暴露其run()的能力（Runaable接口）。

FutureTask類的run()方法

通過上一章的學習，咱們知道execute()方法最後調用的是task的run()方法，上面咱們傳進去的任務，最後被包裝成了FutureTask，也就是說execute()方法最後會調用到FutureTask的run()方法，因此咱們直接看這個方法就能夠了。

public void run() {
    // 狀態不爲NEW，或者修改成當前線程來運行這個任務失敗，則直接返回
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    
    try {
        // 真正的任務
        Callable<V> c = callable;
        // state必須爲NEW時才運行
        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);
    }
}

能夠看到代碼也比較簡單，先作狀態的檢測，再執行任務，最後處理結果或異常。

執行任務這裏沒啥問題，讓咱們看看處理結果或異常的代碼。

protected void setException(Throwable t) {
    // 將狀態從NEW置爲COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 返回值置爲傳進來的異常（outcome爲調用get()方法時返回的）
        outcome = t;
        // 最終的狀態設置爲EXCEPTIONAL
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        // 調用完成方法
        finishCompletion();
    }
}
protected void set(V v) {
    // 將狀態從NEW置爲COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 返回值置爲傳進來的結果（outcome爲調用get()方法時返回的）
        outcome = v;
        // 最終的狀態設置爲NORMAL
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        // 調用完成方法
        finishCompletion();
    }
}

咋一看，這兩個方法彷佛差很少，不一樣的是出去的結果不同且狀態不同，最後都調用了finishCompletion()方法。

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;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }
    // 鉤子方法，子類重寫
    done();
    // 置空任務
    callable = null;        // to reduce footprint
}

整個run()方法總結下來：

（1）FutureTask有一個狀態state控制任務的運行過程，正常運行結束state從NEW->COMPLETING->NORMAL，異常運行結束state從NEW->COMPLETING->EXCEPTIONAL；

（2）FutureTask保存了運行任務的線程runner，它是線程池中的某個線程；

（3）調用者線程是保存在waiters隊列中的，它是何時設置進去的呢？

（4）任務執行完畢，除了設置狀態state變化以外，還要喚醒調用者線程。

調用者線程是何時保存在FutureTask中（waiters）的呢？查看構造方法：

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}

發現並無相關信息，咱們再試想一下，若是調用者不調用get()方法，那麼這種將來任務是否是跟普通任務沒有什麼區別？確實是的哈，因此只有調用get()方法了纔有必要保存調用者線程到FutureTask中。

因此，咱們來看看get()方法中是什麼鬼。

FutureTask類的get()方法

get()方法調用時若是任務未執行完畢，會阻塞直到任務結束。

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    // 若是狀態小於等於COMPLETING，則進入隊列等待
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    // 返回結果（異常）
    return report(s);
}

是否是很清楚，若是任務狀態小於等於COMPLETING，則進入隊列等待。

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();
        }
        // 4. 若是狀態大於COMPLETING了，則跳出循環並返回
        // 這是自旋的出口
        int s = state;
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        // 若是狀態等於COMPLETING，說明任務快完成了，就差設置狀態到NORMAL或EXCEPTIONAL和設置結果了
        // 這時候就讓出CPU，優先完成任務
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        // 1. 若是隊列爲空
        else if (q == null)
            // 初始化隊列（WaitNode中記錄了調用者線程）
            q = new WaitNode();
        // 2. 未進入隊列
        else if (!queued)
            // 嘗試入隊
            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);
        }
        // 3. 阻塞當前線程（調用者線程）
        else
            // 【本文由公從號「彤哥讀源碼」原創】
            LockSupport.park(this);
    }
}

這裏咱們假設調用get()時任務還未執行，也就是其狀態爲NEW，咱們試着按上面標示的一、二、三、4走一遍邏輯：

（1）第一次循環，狀態爲NEW，直接到1處，初始化隊列並把調用者線程封裝在WaitNode中；

（2）第二次循環，狀態爲NEW，隊列不爲空，到2處，讓包含調用者線程的WaitNode入隊；

（3）第三次循環，狀態爲NEW，隊列不爲空，且已入隊，到3處，阻塞調用者線程；

（4）假設過了一會任務執行完畢了，根據run()方法的分析最後會unpark調用者線程，也就是3處會被喚醒；

（5）第四次循環，狀態確定大於COMPLETING了，退出循環並返回；

問題：爲何要在for循環中控制整個流程呢，把這裏的每一步單獨拿出來寫行不行？

答：由於每一次動做都須要從新檢查狀態state有沒有變化，若是拿出去寫也是能夠的，只是代碼會很是冗長。這裏只分析了get()時狀態爲NEW，其它的狀態也能夠自行驗證，都是能夠保證正確的，甚至兩個線程交叉運行（斷點的技巧）。

OK，這裏返回以後，再看看是怎麼處理最終的結果的。

private V report(int s) throws ExecutionException {
    Object x = outcome;
    // 任務正常結束
    if (s == NORMAL)
        return (V)x;
    // 被取消了
    if (s >= CANCELLED)
        throw new CancellationException();
    // 執行異常
    throw new ExecutionException((Throwable)x);
}

還記得前面分析run的時候嗎，任務執行異常時是把異常放在outcome裏面的，這裏就用到了。

（1）若是正常執行結束，則返回任務的返回值；

（2）若是異常結束，則包裝成ExecutionException異常拋出；

經過這種方式，線程中出現的異常也能夠返回給調用者線程了，不會像執行普通任務那樣調用者是不知道任務執行到底有沒有成功的。

其它

FutureTask除了能夠獲取任務的返回值之外，還可以取消任務的執行。

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}

這裏取消任務是經過中斷執行線程來處理的，有興趣的同窗能夠本身分析一下。

回答開篇

若是這裏把future.get()放到for循環裏面，時間大概是多少？

答：大概會是5秒多一點，由於每提交一個任務，都要阻塞調用者線程直到任務執行完畢，每一個任務執行都是1秒多，因此總時間就是5秒多點。

總結

（1）將來任務是經過把普通任務包裝成FutureTask來實現的。

（2）經過FutureTask不只可以獲取任務執行的結果，還有感知到任務執行的異常，甚至還能夠取消任務；

（3）AbstractExecutorService中定義了不少模板方法，這是一種很重要的設計模式；

（4）FutureTask其實就是典型的異常調用的實現方式，後面咱們學習到Netty、Dubbo的時候還會見到這種設計思想的。

彩蛋

RPC框架中異步調用是怎麼實現的？

答：RPC框架經常使用的調用方式有同步調用、異步調用，其實它們本質上都是異步調用，它們就是用FutureTask的方式來實現的。

通常地，經過一個線程（咱們叫做遠程線程）去調用遠程接口，若是是同步調用，則直接讓調用者線程阻塞着等待遠程線程調用的結果，待結果返回了再返回；若是是異步調用，則先返回一個將來能夠獲取到遠程結果的東西FutureXxx，固然，若是這個FutureXxx在遠程結果返回以前調用了get()方法同樣會阻塞着調用者線程。

有興趣的同窗能夠先去預習一下dubbo的異步調用（它是把Future扔到RpcContext中的）。

歡迎關注個人公衆號「彤哥讀源碼」，查看更多源碼系列文章, 與彤哥一塊兒暢遊源碼的海洋。