Java 中「Future」接口詳解
一、背景
在系統中，異步執行任務，是很常見的功能邏輯，但是在不同的場景中，又存在很多細節差異；
有的任務只強調「執行過程」，並不需要追溯任務自身的「執行結果」，這裏並不是指對系統和業務產生的效果，比如定時任務、消息隊列等場景；
但是有些任務即強調「執行過程」，又需要追溯任務自身的「執行結果」，在流程中依賴某個異步結果，判斷流程是否中斷，比如「並行」處理；
【串行處理】整個流程按照邏輯逐步推進，如果出現異常會導致流程中斷；
【並行處理】主流程按照邏輯逐步推進，其他「異步」交互的流程執行完畢後，將結果返回到主流程，如果「異步」流程異常，會影響部分結果；
此前在《「訂單」業務》的內容中，聊過關於「串行」和「並行」的應用對比，即在訂單詳情的加載過程中，通過「並行」的方式讀取：商品、商戶、訂單、用戶等信息，提升接口的響應時間；
二、Future 接口
1、入門案例
異步是對流程的解耦，但是有的流程中又依賴異步執行的最終結果，此時就可以使用「Future」接口來達到該目的，先來看一個簡單的入門案例；
public class ServerTask implements Callable<Integer> {
@Override
public Integer call() throws Exception {
Thread.sleep(2000);
return 3;
}
}
public class FutureBase01 {
public static void main(String[] args) throws Exception {
TimeInterval timer = DateUtil.timer();
// 線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 批量任務
List<ServerTask> serverTasks = new ArrayList<>() ;
for (int i=0;i<3;i++){
serverTasks.add(new ServerTask());
}
List<Future<Integer>> taskResList = executor.invokeAll(serverTasks) ;
// 結果輸出
for (Future<Integer> intFuture:taskResList){
System.out.println(intFuture.get());
}
// 耗時統計
System.out.println("timer...interval = "+timer.interval());
}
}這裏模擬一個場景，以線程池批量執行異步任務，在任務內線程休眠 2 秒，以並行的方式最終獲取全部結果，只耗時 2 秒多一點，如果串行的話耗時肯定超過 6 秒；
2、Future 接口
Future 表示異步計算的結果，提供了用於檢查計算是否完成、等待計算完成、以及檢索計算結果的方法。
【核心方法】
-
get()：等待任務完成，獲取執行結果，如果任務取消會拋出異常； -
get(long timeout, TimeUnit unit)：指定等待任務完成的時間，等待超時會拋出異常； -
isDone()：判斷任務是否完成； -
isCancelled()：判斷任務是否被取消； -
cancel(boolean mayInterruptIfRunning)：嘗試取消此任務的執行，如果任務已經完成、已經取消或由於其他原因無法取消，則此嘗試將失敗；
【基礎用法】
public class FutureBase02 {
public static void main(String[] args) throws Exception {
// 線程池執行任務
ExecutorService executor = Executors.newFixedThreadPool(3);
FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
@Override
public String call() throws Exception {
Thread.sleep(3000);
return "task...OK";
}
}) ;
executor.execute(futureTask);
// 任務信息獲取
System.out.println("是否完成："+futureTask.isDone());
System.out.println("是否取消："+futureTask.isCancelled());
System.out.println("獲取結果："+futureTask.get());
System.out.println("嘗試取消："+futureTask.cancel(Boolean.TRUE));
}
}【FutureTask】
Future 接口的基本實現類，提供了計算的啓動和取消、查詢計算是否完成以及檢索計算結果的方法；
在「FutureTask」類中，可以看到線程異步執行任務時，其中的核心狀態轉換，以及最終結果寫出的方式；
雖然「Future」從設計上，實現了異步計算的結果獲取，但是通過上面的案例也可以發現，流程的主線程在執行get()方法時會阻塞，直到最終獲取結果，顯然對於程序來說並不友好；
在JDK1.8提供「CompletableFuture」類，對「Future」進行優化和擴展；
三、CompletableFuture 類
1、基礎說明
「CompletableFuture」類提供函數編程的能力，可以通過回調的方式處理計算結果，並且支持組合操作，提供很多方法來實現異步編排，降低異步編程的複雜度；
「CompletableFuture」實現「Future」和「CompletionStage」兩個接口；
-
Future：表示異步計算的結果；
-
CompletionStage：表示異步計算的一個步驟，當一個階段計算完成時，可能會觸發其他階段，即步驟可能由其他 CompletionStage 觸發；
【入門案例】
public class CompletableBase01 {
public static void main(String[] args) throws Exception {
// 線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 任務執行
CompletableFuture<String> cft = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "Res...OK";
}, executor);
// 結果輸出
System.out.println(cft.get());
}
}2、核心方法
2.1 實例方法
public class Completable01 {
public static void main(String[] args) throws Exception {
// 線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 1、創建未完成的CompletableFuture，通過complete()方法完成
CompletableFuture<Integer> cft01 = new CompletableFuture<>() ;
cft01.complete(99) ;
// 2、創建已經完成CompletableFuture，並且給定結果
CompletableFuture<String> cft02 = CompletableFuture.completedFuture("given...value");
// 3、有返回值，默認ForkJoinPool線程池
CompletableFuture<String> cft03 = CompletableFuture.supplyAsync(() -> {return "OK-3";});
// 4、有返回值，採用Executor自定義線程池
CompletableFuture<String> cft04 = CompletableFuture.supplyAsync(() -> {return "OK-4";},executor);
// 5、無返回值，默認ForkJoinPool線程池
CompletableFuture<Void> cft05 = CompletableFuture.runAsync(() -> {});
// 6、無返回值，採用Executor自定義線程池
CompletableFuture<Void> cft06 = CompletableFuture.runAsync(()-> {}, executor);
}
}2.2 計算方法
public class Completable02 {
public static void main(String[] args) throws Exception {
// 線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK";
},executor);
// 1、計算完成後，執行後續處理
// cft01.whenComplete((res, ex) -> System.out.println("Result："+res+"；Exe："+ex));
// 2、觸發計算，如果沒有完成，則get設定的值，如果已完成，則get任務返回值
// boolean completeFlag = cft01.complete("given...value");
// if (completeFlag){
// System.out.println(cft01.get());
// } else {
// System.out.println(cft01.get());
// }
// 3、開啓新CompletionStage，重新獲取線程執行任務
cft01.whenCompleteAsync((res, ex) -> System.out.println("Result："+res+"；Exe："+ex),executor);
}
}2.3 結果獲取方法
public class Completable03 {
public static void main(String[] args) throws Exception {
// 線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "Res...OK";
},executor);
// 1、阻塞直到獲取結果
// System.out.println(cft01.get());
// 2、設定超時的阻塞獲取結果
// System.out.println(cft01.get(4, TimeUnit.SECONDS));
// 3、非阻塞獲取結果，如果任務已經完成，則返回結果，如果任務未完成，返回給定的值
// System.out.println(cft01.getNow("given...value"));
// 4、get獲取拋檢查異常，join獲取非檢查異常
System.out.println(cft01.join());
}
}2.4 任務編排方法
public class Completable04 {
public static void main(String[] args) throws Exception {
// 線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("OK-1");
return "OK";
},executor);
// 1、cft01任務執行完成後，執行之後的任務，此處不關注cft01的結果
// cft01.thenRun(() -> System.out.println("task...run")) ;
// 2、cft01任務執行完成後，執行之後的任務，可以獲取cft01的結果
// cft01.thenAccept((res) -> {
// System.out.println("cft01："+res);
// System.out.println("task...run");
// });
// 3、cft01任務執行完成後，執行之後的任務，獲取cft01的結果，並且具有返回值
// CompletableFuture<Integer> cft02 = cft01.thenApply((res) -> {
// System.out.println("cft01："+res);
// return 99 ;
// });
// System.out.println(cft02.get());
// 4、順序執行cft01、cft02
// CompletableFuture<String> cft02 = cft01.thenCompose((res) -> CompletableFuture.supplyAsync(() -> {
// System.out.println("cft01："+res);
// return "OK-2";
// }));
// cft02.whenComplete((res,ex) -> System.out.println("Result："+res+"；Exe："+ex));
// 5、對比任務的執行效率，由於cft02先完成，所以取cft02的結果
// CompletableFuture<String> cft02 = cft01.applyToEither(CompletableFuture.supplyAsync(() -> {
// System.out.println("run...cft02");
// try {
// Thread.sleep(3000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// return "OK-2";
// }),(res) -> {
// System.out.println("either...result：" + res);
// return res;
// });
// System.out.println("finally...result：" + cft02.get());
// 6、兩組任務執行完成後，對結果進行合併
// CompletableFuture<String> cft02 = CompletableFuture.supplyAsync(() -> "OK-2") ;
// String finallyRes = cft01.thenCombine(cft02,(res1,res2) -> {
// System.out.println("res1："+res1+"；res2："+res2);
// return res1+"；"+res2 ;
// }).get();
// System.out.println(finallyRes);
CompletableFuture<String> cft02 = CompletableFuture.supplyAsync(() -> {
System.out.println("OK-2");
return "OK-2";
}) ;
CompletableFuture<String> cft03 = CompletableFuture.supplyAsync(() -> {
System.out.println("OK-3");
return "OK-3";
}) ;
// 7、等待批量任務執行完返回
// CompletableFuture.allOf(cft01,cft02,cft03).get();
// 8、任意一個任務執行完即返回
System.out.println("Sign："+CompletableFuture.anyOf(cft01,cft02,cft03).get());
}
}2.5 異常處理方法
public class Completable05 {
public static void main(String[] args) throws Exception {
// 線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
if (1 > 0){
throw new RuntimeException("task...exception");
}
return "OK";
},executor);
// 1、捕獲cft01的異常信息，並提供返回值
String finallyRes = cft01.thenApply((res) -> {
System.out.println("cft01-res：" + res);
return res;
}).exceptionally((ex) -> {
System.out.println("cft01-exe：" + ex.getMessage());
return "error" ;
}).get();
System.out.println("finallyRes="+finallyRes);
CompletableFuture<String> cft02 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK-2";
},executor);
// 2、如果cft02未完成，則get時拋出指定異常信息
boolean exeFlag = cft02.completeExceptionally(new RuntimeException("given...exception"));
if (exeFlag){
System.out.println(cft02.get());
} else {
System.out.println(cft02.get());
}
}
}3、線程池問題
-
在實踐中，通常不使用
ForkJoinPool#commonPool()公共線程池，會出現線程競爭問題，從而形成系統瓶頸； -
在任務編排中，如果出現依賴情況或者父子任務，儘量使用多個線程池，從而避免任務請求同一個線程池，規避死鎖情況發生；
四、CompletableFuture 原理
1、核心結構
在分析「CompletableFuture」其原理之前，首先看一下涉及的核心結構；
【CompletableFuture】
在該類中有兩個關鍵的字段：「result」存儲當前 CF 的結果，「stack」代表棧頂元素，即當前 CF 計算完成後會觸發的依賴動作；從上面案例中可知，依賴動作可以沒有或者有多個；
【Completion】
依賴動作的封裝類；
【UniCompletion】
繼承 Completion 類，一元依賴的基礎類，「executor」指線程池，「dep」指依賴的計算，「src」指源動作；
【BiCompletion】
繼承 UniCompletion 類，二元或者多元依賴的基礎類，「snd」指第二個源動作；
2、零依賴
顧名思義，即各個 CF 之間不產生依賴關係；
public class DepZero {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(()-> "OK-1",executor);
CompletableFuture<String> cft2 = CompletableFuture.supplyAsync(()-> "OK-2",executor);
System.out.println(cft1.get()+";"+cft2.get());
}
}3、一元依賴
即 CF 之間的單個依賴關係；這裏使用「thenApply」方法演示，爲了看到效果，使「cft1」長時間休眠，斷點查看「stack」結構；
public class DepOne {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK-1";
},executor);
CompletableFuture<String> cft2 = cft1.thenApply(res -> {
System.out.println("cft01-res"+res);
return "OK-2" ;
});
System.out.println("cft02-res"+cft2.get());
}
}斷點截圖：
原理分析：
觀察者 Completion 註冊到「cft1」，註冊時會檢查計算是否完成，未完成則觀察者入棧，當「cft1」計算完成會彈棧；已完成則直接觸發觀察者；
可以調整斷點代碼，讓「cft1」先處於完成狀態，再查看其運行時結構，從而分析完整的邏輯；
4、二元依賴
即一個 CF 同時依賴兩個 CF；這裏使用「thenCombine」方法演示；爲了看到效果，使「cft1、cft2」長時間休眠，斷點查看「stack」結構；
public class DepTwo {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK-1";
},executor);
CompletableFuture<String> cft2 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK-2";
},executor);
// cft3 依賴 cft1和cft2 的計算結果
CompletableFuture<String> cft3 = cft1.thenCombine(cft2,(res1,res2) -> {
System.out.println("cft01-res:"+res1);
System.out.println("cft02-res:"+res2);
return "OK-3" ;
});
System.out.println("cft03-res:"+cft3.get());
}
}斷點截圖：
原理分析：
在「cft1」和「cft2」未完成的狀態下，嘗試將 BiApply 壓入「cft1」和「cft2」兩個棧中，任意 CF 完成時，會嘗試觸發觀察者，觀察者檢查「cft1」和「cft2」是否都完成，如果完成則執行；
5、多元依賴
即一個 CF 同時依賴多個 CF；這裏使用「allOf」方法演示；爲了看到效果，使「cft1、cft2、cft3」長時間休眠，斷點查看「stack」結構；
public class DepMore {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK-1";
},executor);
CompletableFuture<String> cft2 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK-2";
},executor);
CompletableFuture<String> cft3 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK-3";
},executor);
// cft4 依賴 cft1和cft2和cft3 的計算結果
CompletableFuture<Void> cft4 = CompletableFuture.allOf(cft1,cft2,cft3);
CompletableFuture<String> finallyRes = cft4.thenApply(tm -> {
System.out.println("cft01-res:"+cft1.join());
System.out.println("cft02-res:"+cft2.join());
System.out.println("cft03-res:"+cft3.join());
return "OK-4";
});
System.out.println("finally-res:"+finallyRes.get());
}
}斷點截圖：
原理分析：
多元依賴的回調方法除了「allOf」還有「anyOf」，其實現原理都是將依賴的多個 CF 補全爲平衡二叉樹，從斷點圖可知會按照樹的層級處理，核心結構參考二元依賴即可；
五、參考源碼
編程文檔：
https://gitee.com/cicadasmile/butte-java-note
應用倉庫：
https://gitee.com/cicadasmile/butte-flyer-parent本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/9lv36J8x29ilj6R7aOWirw