CompletableFuture非同步編排
- 2020 年 3 月 30 日
- 筆記
什麼是CompletableFuture
CompletableFuture是JDK8提供的Future增強類。CompletableFuture非同步任務執行執行緒池，默認是把非同步任務都放在ForkJoinPool中執行。
在這種方式中，主執行緒不會被阻塞，不需要一直等到子執行緒完成。主執行緒可以並行的執行其他任務。
Future存在的問題
Future實際採用FutureTask實現，該對象相當於是消費者和生產者的橋樑，消費者通過 FutureTask 存儲任務的處理結果，更新任務的狀態：未開始、正在處理、已完成等。而生產者拿到的 FutureTask 被轉型為 Future 介面，可以阻塞式獲取任務的處理結果，非阻塞式獲取任務處理狀態。
使用
runAsync 和 supplyAsync方法
CompletableFuture 提供了四個靜態方法來創建一個非同步操作。
public static CompletableFuture<Void> runAsync(Runnable runnable) public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) 沒有指定Executor的方法會使用ForkJoinPool.commonPool() 作為它的執行緒池執行非同步程式碼。如果指定執行緒池，則使用指定的執行緒池運行。以下所有的方法都類同。
- runAsync方法不支援返回值。
- supplyAsync可以支援返回值。
計算完成時回調方法
當CompletableFuture的計算結果完成，或者拋出異常的時候，可以執行特定的Action。主要是下面的方法：
public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action); public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action); public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor); public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn); whenComplete可以處理正常和異常的計算結果，exceptionally處理異常情況。BiConsumer<? super T,? super Throwable>可以定義處理業務
whenComplete 和 whenCompleteAsync 的區別：
whenComplete：是執行當前任務的執行緒執行繼續執行 whenComplete 的任務。
whenCompleteAsync：是執行把 whenCompleteAsync 這個任務繼續提交給執行緒池來進行執行。
方法不以Async結尾，意味著Action使用相同的執行緒執行，而Async可能會使用其他執行緒執行（如果是使用相同的執行緒池，也可能會被同一個執行緒選中執行）
程式碼示例：
public class CompletableFutureDemo { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture future = CompletableFuture.supplyAsync(new Supplier<Object>() { @Override public Object get() { System.out.println(Thread.currentThread().getName() + "t completableFuture"); int i = 10 / 0; return 1024; } }).whenComplete(new BiConsumer<Object, Throwable>() { @Override public void accept(Object o, Throwable throwable) { System.out.println("-------o=" + o.toString()); System.out.println("-------throwable=" + throwable); } }).exceptionally(new Function<Throwable, Object>() { @Override public Object apply(Throwable throwable) { System.out.println("throwable=" + throwable); return 6666; } }); System.out.println(future.get()); } } handle 方法
handle 是執行任務完成時對結果的處理。
handle 是在任務完成後再執行，還可以處理異常的任務。
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn); public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn); public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor); 執行緒串列化方法
thenApply 方法：當一個執行緒依賴另一個執行緒時，獲取上一個任務返回的結果，並返回當前任務的返回值。
thenAccept方法：消費處理結果。接收任務的處理結果，並消費處理，無返回結果。
thenRun方法：只要上面的任務執行完成，就開始執行thenRun，只是處理完任務後，執行 thenRun的後續操作
帶有Async默認是非同步執行的。這裡所謂的非同步指的是不在當前執行緒內執行。
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) public CompletionStage<Void> thenAccept(Consumer<? super T> action); public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action); public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor); public CompletionStage<Void> thenRun(Runnable action); public CompletionStage<Void> thenRunAsync(Runnable action); public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor); Function<? super T,? extends U>
T：上一個任務返回結果的類型
U：當前任務的返回值類型
程式碼演示：
public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new Supplier<Integer>() { @Override public Integer get() { System.out.println(Thread.currentThread().getName() + "t completableFuture"); //int i = 10 / 0; return 1024; } }).thenApply(new Function<Integer, Integer>() { @Override public Integer apply(Integer o) { System.out.println("thenApply方法，上次返回結果：" + o); return o * 2; } }).whenComplete(new BiConsumer<Integer, Throwable>() { @Override public void accept(Integer o, Throwable throwable) { System.out.println("-------o=" + o); System.out.println("-------throwable=" + throwable); } }).exceptionally(new Function<Throwable, Integer>() { @Override public Integer apply(Throwable throwable) { System.out.println("throwable=" + throwable); return 6666; } }).handle(new BiFunction<Integer, Throwable, Integer>() { @Override public Integer apply(Integer integer, Throwable throwable) { System.out.println("handle o=" + integer); System.out.println("handle throwable=" + throwable); return 8888; } }); System.out.println(future.get()); } 兩任務組合 – 都要完成
兩個任務必須都完成，觸發該任務。
thenCombine：組合兩個future，獲取兩個future的返回結果，並返回當前任務的返回值
thenAcceptBoth：組合兩個future，獲取兩個future任務的返回結果，然後處理任務，沒有返回值。
runAfterBoth：組合兩個future，不需要獲取future的結果，只需兩個future處理完任務後，處理該任務。
public <U,V> CompletableFuture<V> thenCombine( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn); public <U,V> CompletableFuture<V> thenCombineAsync( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn); public <U,V> CompletableFuture<V> thenCombineAsync( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor); public <U> CompletableFuture<Void> thenAcceptBoth( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action); public <U> CompletableFuture<Void> thenAcceptBothAsync( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action); public <U> CompletableFuture<Void> thenAcceptBothAsync( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor); public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action); public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action); public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor); 測試案例：
public static void main(String[] args) { CompletableFuture.supplyAsync(() -> { return "hello"; }).thenApplyAsync(t -> { return t + " world!"; }).thenCombineAsync(CompletableFuture.completedFuture(" CompletableFuture"), (t, u) -> { return t + u; }).whenComplete((t, u) -> { System.out.println(t); }); } 輸出：hello world! CompletableFuture
兩任務組合 – 一個完成
當兩個任務中，任意一個future任務完成的時候，執行任務。
applyToEither：兩個任務有一個執行完成，獲取它的返回值，處理任務並有新的返回值。
acceptEither：兩個任務有一個執行完成，獲取它的返回值，處理任務，沒有新的返回值。
runAfterEither：兩個任務有一個執行完成，不需要獲取future的結果，處理任務，也沒有返回值。
public <U> CompletableFuture<U> applyToEither( CompletionStage<? extends T> other, Function<? super T, U> fn); public <U> CompletableFuture<U> applyToEitherAsync( CompletionStage<? extends T> other, Function<? super T, U> fn); public <U> CompletableFuture<U> applyToEitherAsync( CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor); public CompletableFuture<Void> acceptEither( CompletionStage<? extends T> other, Consumer<? super T> action); public CompletableFuture<Void> acceptEitherAsync( CompletionStage<? extends T> other, Consumer<? super T> action); public CompletableFuture<Void> acceptEitherAsync( CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor); public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action); public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action); public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor); 多任務組合
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs); public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs); allOf：等待所有任務完成
anyOf：只要有一個任務完成
public static void main(String[] args) { List<CompletableFuture> futures = Arrays.asList(CompletableFuture.completedFuture("hello"), CompletableFuture.completedFuture(" world!"), CompletableFuture.completedFuture(" hello"), CompletableFuture.completedFuture("java!")); final CompletableFuture<Void> allCompleted = CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{})); allCompleted.thenRun(() -> { futures.stream().forEach(future -> { try { System.out.println("get future at:"+System.currentTimeMillis()+", result:"+future.get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } }); }); } 
