向線程池提交Callable任務,會建立一個新線程(執行任務的線程)去執行這個Callable任務,可是經過Future#get獲取任務的執行結果是在提交任務的調用者線程中,那問題一:調用者線程如何獲取執行任務的線程的結果?java
在JDK中,有2種類型的任務,Runnable和Callable,可是具體到線程池執行任務的java.util.concurrent.ThreadPoolExecutor#execute(Runnable)
方法,它只接收Runnable任務,那問題二:Callable任務是提交給線程池後是如何執行的呢?promise
import java.util.concurrent.*; public class FutureTest { public static void main(String[] args) { Callable<Integer> callable = new Callable<Integer>() { @Override public Integer call() throws Exception { //sleep 是爲了調試方便 TimeUnit.SECONDS.sleep(4); return 3; } }; //建立一個 ThreadPoolExecutor 對象 ExecutorService executorService = Executors.newFixedThreadPool(1); Future<Integer> future = executorService.submit(callable); try { Integer i = future.get(); System.out.println(i); } catch (Exception e) { System.out.println(e); } } }
Future<Integer> future = executorService.submit(callable);
less
//java.util.concurrent.AbstractExecutorService#submit(java.util.concurrent.Callable<T>) public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); //FutureTask實際上是個RunnableFuture, RunnableFuture實際上是個Runnable //重點是: Runnable#run方法的執行,其實就是 FutureTask#run方法的執行!!! RunnableFuture<T> ftask = newTaskFor(task); //java.util.concurrent.ThreadPoolExecutor#execute execute(ftask); return ftask; }
RunnableFuture<T> ftask = newTaskFor(task);
jvm
//java.util.concurrent.AbstractExecutorService#newTaskFor(java.util.concurrent.Callable<T>) protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { return new FutureTask<T>(callable); }
當submit一個Callable任務時,會生成一個RunnableFuture接口對象,默認狀況下 RunnableFuture對象是一個FutureTask對象。看java.util.concurrent.AbstractExecutorService
類的源碼註釋:咱們也能夠重寫 newTaskFor 方法生成咱們本身的 RunnableFuture。一個具體的示例可參考ES源碼org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor#newTaskFor(java.util.concurrent.Callable<T>)
,它就重寫了 newTaskFor 方法,實現了執行優先級任務時,獲取任務執行結果的邏輯。elasticsearch
the implementation of submit(Runnable) creates an associated RunnableFuture that is executed and returned. Subclasses may override the newTaskFor methods to return RunnableFuture implementations other than FutureTaskide
而後再來看FutureTask這個類的run()方法:java.util.concurrent.FutureTask#run
,它會觸發執行咱們定義的Callable#call()方法。搞清楚java.util.concurrent.FutureTask#run方法是怎麼被調用的,就搞清楚了線程池執行Callable任務的原理。該方法主要是作了2件事:oop
set(result)
java.util.concurrent.AbstractExecutorService#submit(java.lang.Runnable) 中execute(ftask)
提交任務(注意:FutureTask implements Runnable)this
ThreadPoolExecutor是AbstractExecutorService具體實現類,所以最終會執行到:java.util.concurrent.ThreadPoolExecutor#execute提交任務。線程
//java.util.concurrent.ThreadPoolExecutor#execute, 重點看addWorker()實現 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); }
java.util.concurrent.ThreadPoolExecutor#addWorker 有2行代碼很關鍵:調試
//java.util.concurrent.ThreadPoolExecutor#addWorker try { w = new Worker(firstTask);//關鍵代碼1, firstTask 本質上是 FutureTask對象 final Thread t = w.thread; if (t != null) { //...省略非關鍵代碼 if (workerAdded) { t.start();//關鍵代碼 2 workerStarted = true; } } }
w = new Worker(firstTask)
建立一個新線程!把Worker做爲this對象傳進去,由於Worker implements Runnable,而且實現了java.lang.Runnable#run方法。
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask;// this.thread = getThreadFactory().newThread(this); }
這意味着啥?執行java.lang.Runnable#run 就會去真正地執行 java.util.concurrent.ThreadPoolExecutor.Worker#run,那麼java.lang.Runnable#run是被誰調用的呢?
聰明的你必定知道了,new Thread(Runnable).start()
執行時,會由jvm去自動調用java.lang.Runnable#run
因此,上面java.util.concurrent.ThreadPoolExecutor#addWorker 中的關鍵代碼2 t.start();
,觸發了java.util.concurrent.ThreadPoolExecutor.Worker#run的調用。
java.util.concurrent.ThreadPoolExecutor.Worker#run
裏面只是調用了runWoker(this)
而已。
//java.util.concurrent.ThreadPoolExecutor.Worker#run /** Delegates main run loop to outer runWorker. */ public void run() { runWorker(this); }
重點來了!再跟進去看看runWoker是何方神聖:
//java.util.concurrent.ThreadPoolExecutor#runWorker final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask;//task 其實是FutureTask類型的對象 w.firstTask = null; try { while (task != null || (task = getTask()) != null) { //省略一些 非關鍵代碼.... try { beforeExecute(wt, task);// try { //重點代碼!觸發 java.util.concurrent.FutureTask#run 執行 task.run(); afterExecute(task, null); } catch (Throwable ex) { //去看看afterExecute方法註釋,不管線程執行過程當中是否拋異常,afterExecute()都會 執行,看了源碼,明白爲何是這樣了,由於catch異常處理裏面會執行afterExecute afterExecute(task, ex); throw ex; } } finally { task = null; w.completedTasks++; w.unlock(); } }
看懂了java.util.concurrent.ThreadPoolExecutor#runWorker
幾乎就明白線程池執行任務時的beforeExecute、afterExecute方法的所起的做用了(好比常常在afterExecute方法裏面作一些線程池任務運行時間的統計工做)。
總結如下點:
Callable任務被submit時,會生成一個FutureTask對象,封裝Callable,在FutureTask的run方法裏面執行Callable#call方法,而且調用java.util.concurrent.FutureTask#set
設置Callable任務的執行結果(結果保存在一個FutureTask的Object類型的實例變量裏面:private Object outcome;
)。
Future<Integer> future = executorService.submit(callable);
返回一個Future,它其實是一個FutureTask對象,經過java.util.concurrent.FutureTask#get()
獲取Callable任務的執行結果。
java.util.concurrent.FutureTask#run
方法是由java.util.concurrent.ThreadPoolExecutor#runWorker
觸發調用的;而java.util.concurrent.ThreadPoolExecutor#runWorker
又是由java.util.concurrent.ThreadPoolExecutor.Worker#run
觸發調用的;而java.util.concurrent.ThreadPoolExecutor.Worker#run
又是由java.util.concurrent.ThreadPoolExecutor#addWorker
裏面的t.start();
這條語句觸發調用的;而t.start();
會觸發Runnable#run
方法的執行。這就是前面提到的這個原理:new Thread(Runnable).start()
會由jvm來調用Runnable#run。具體可參考:
用一個詞表示就是多態。用一張圖表示就是:
繼承 ThreadPoolExecutor 實現自定義的線程池時,可重寫 afterExecute()方法作一些異常處理邏輯的實現,無論任務正常執行完成、仍是拋出異常,都會調用afterExecute(),具體可看JDK源碼關於ThreadPoolExecutor#runWorker方法的註釋。有興趣可研究下ES SEARCH線程池源碼就使用afterExecute來統計提交給線程池的每一個任務的等待時間、執行時間,從而根據Little's law 自動調整線程池任務隊列的長度:org.elasticsearch.common.util.concurrent.QueueResizingEsThreadPoolExecutor#afterExecute
最後,想說的是:Callable任務,到ThreadPoolExecutor線程池執行 層面,它其實是一個Runnable任務在執行。由於,ExecutorService submit Callable時,實際上是將Callable封裝到FutureTask/RunnableFuture中,而RunnableFuture implements Runnable,所以能夠提交給線程池的java.util.concurrent.ThreadPoolExecutor#execute(Runnable command)
執行,這就回答了本文開頭提出的第二個問題。
//java.util.concurrent.RunnableFuture public interface RunnableFuture<V> extends Runnable, Future<V> { /** * Sets this Future to the result of its computation * unless it has been cancelled. */ void run(); }
用一張圖表示就是:
Callable任務的設置與獲取,則都是在FutureTask這個層面上完成,把Callable封裝到FutureTask中,而FutureTask implements Runnable,從而轉化成ThreadPoolExecutor#execute執行Runnable任務。
java.util.concurrent.FutureTask 的private volatile int state;
變量:
//java.util.concurrent.FutureTask#run 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 { //Callable#call執行成功, ran=true result = c.call(); ran = true; } catch (Throwable ex) { result = null; ran = false; setException(ex); } //ran=true,纔會設置Callable任務的執行結果 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); } }
set方法設置Callable任務的執行結果時,會修改 FutureTask的 state 實例變量的值!
//java.util.concurrent.FutureTask#set protected void set(V v) { if (STATE.compareAndSet(this, NEW, COMPLETING)) { outcome = v; STATE.setRelease(this, NORMAL); // final state finishCompletion(); } }
而java.util.concurrent.FutureTask#get()
方法,也正是經過檢查 state 的值,來肯定是否可以拿到Callable任務的執行結果。
//java.util.concurrent.FutureTask#get() public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) //若是 state 不是在 NORMAL 狀態,FutureTask#get()就會阻塞 //這就是 java.util.concurrent.Future#get() 阻塞的緣由 s = awaitDone(false, 0L);//這裏面會調用:Thread.yield()、LockSupport.park(this) return report(s); }
java.util.concurrent.FutureTask#awaitDone
//java.util.concurrent.FutureTask#awaitDone private int awaitDone(boolean timed, long nanos) throws InterruptedException { WaitNode q = null; //省略一些無關代碼... for (;;) {//for循環一直檢查任務的運行狀態....直到能夠"結束" int s = state; //state的值大於 COMPLETING 說明已經有Callable任務的結果了 //java.util.concurrent.FutureTask#set 設置了Callable任務的結果,修改了state的值 if (s > COMPLETING) { if (q != null) q.thread = null; return s; } //COMPLETING 任務的運行狀態是:正在執行中 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();//掛起獲取執行結果的線程(這就是Futur#get阻塞的緣由) 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); //java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit)超時阻塞的實現原理 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); } }
總結一下:經過 state變量來判斷Callable任務的執行結果是否已經生成。若是已經生成了執行結果,那麼 java.util.concurrent.FutureTask#set
會把結果放到private Object outcome;
outcome這個變量中。而後設置state的值爲NORMAL,那麼java.util.concurrent.FutureTask#get()
經過檢查 state 的值,就能拿到執行結果了,固然了,若是執行結果還未生成,java.util.concurrent.FutureTask#awaitDone
就會致使 get 阻塞。
最後的最後,留一個問題:因爲JDK裏面Future#get都是阻塞的,那有沒有什麼方法使得獲取 Callable 任務的執行結果不阻塞?
看看Netty的源碼?借鑑一下Listener回調機制。哈哈……