CompleteFuture實現簡單的任務編排實踐
CompleteFuture實現簡單的任務編排實踐
一：前言
CompleteFuture是java8 新提供的API，是對函數式編程思想的體現，提供了很多的對於函數式編程支持。不止有同步處理功能，還有異步處理能力。
通過函數式編程可以實現線程的簡單任務編排。高效，整潔實現多線程異步編程。
二：詳細介紹
CompleteFuture 提供的API中以ansy結尾的都是異步處理的。
-
異步執行任務，並返回結果：
supplyAsync異步處理，並返回結果，默認使用ForkJoinPool.commonPool()線程池，同時提供支持自定義線程池的API。CompletableFuture.supplyAsync(() -> "HELLO"); // 自定義線程池 CompletableFuture.supplyAsync(()->"hello",ES);- 異步執行任務，不返回結果：
runAsync
CompletableFuture.runAsync(() -> System.out.println("HELLO WORLD !")); CompletableFuture.runAsync(() -> System.out.println("HELLO WORLD !"),ES);- 依賴單一階段：
thenApply thenApplyAsync
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "HELLO") .thenApply(a -> return a + " lili!"; });- 組合與撰寫：
thenCompose()，thenCombine()，thenCombineAsync.
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "hello") .thenCompose(res -> CompletableFuture.supplyAsync(() -> res + " lili")) .thenCompose(res -> CompletableFuture.supplyAsync(() -> res + " lucy")); // 執行結果： =====> hello lili lucy // mian線程下同步執行。CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "hello") .thenCompose(res -> CompletableFuture.supplyAsync(() -> res + " lili")) .thenCompose(res -> CompletableFuture.supplyAsync(() -> res + " lucy")) .thenCombineAsync(CompletableFuture.supplyAsync(() -> " how are you!"), (a, b) -> a + b); log.info("=====> {}", f1.get()); // 執行結果： =====> hello lili lucy how are you!- 依賴兩個任務中的一個：
applyToEither()，那個任務先結束，就依賴那個任務。
CompletableFuture<String> voidCompletableFuture = CompletableFuture.supplyAsync(() -> { try {TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace();} return "lucy"; }).applyToEither(CompletableFuture.supplyAsync(() -> { try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();} return "lili"; }), a -> "hello " + a); log.info("ret ====> {}",voidCompletableFuture.get()); // 執行結果： ret ====> hello lili 如果下面sleep改成3s，執行結果：ret ====> hello lucy- 消費型，依賴單階段：
thenAccept()，thenAcceptAsync()
CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> "hello") .thenAcceptAsync(a -> { a = a + " lucy !"; log.info("ret ======> {}", a); }); log.info(" ======== end ========================"); // 執行結果：ret ======> hello lucy ! 而且是異步的，不會阻塞主線程，下面的end是先打印出來的- 消費型，依賴兩個任務都完成：
thenAcceptBoth()，thenAcceptBothAsync()
CompletableFuture.supplyAsync(() -> "hello") .thenAcceptBoth(CompletableFuture.supplyAsync(() -> " lili"), (a, b) -> { try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } log.info("=======>{}", a + b); }); // 執行結果：=======>hello lili- 消費型：
acceptEither()依賴兩個任務中先執行結束的那個
CompletableFuture.supplyAsync(() -> { try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } return "lucy"; }).acceptEither(CompletableFuture.supplyAsync(() -> "lili"), a -> { log.info("hello {}", a); }); // 執行結果：hello lili- 消費型，無論正常，還是異常都會消費處理，而且不會吞掉異常
whenComplete()，whenCompleteAsync()
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { if (ThreadLocalRandom.current().nextInt(2) < 2) { throw new RuntimeException("error"); } return "hello"; }).whenComplete((a, e) -> { log.info("ret -> {}", a + " lili!"); log.error("error", e); }); log.info("future.get()-->{}", future.get()); // 執行結果：ret -> null lili! 而且打印兩次錯誤日誌，一次是log打印，一次是get的時候。- 產出型，無論正常還是異常都是處理，並返回結果。
handle，handleAsync
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "hello") .handle((a, e) -> a + " lili!"); log.info("ret ==> {}", future.get()); // 執行結果：ret ==> hello lili!- 產出型，異常時候進行處理，併產出，有點像try-catch()，
exceptionally()
CompletableFuture<Object> f = CompletableFuture.supplyAsync(() -> "Hello") .thenApplyAsync(res -> res + " World") .thenApplyAsync( res -> { throw new RuntimeException(" test has error"); // return res + "!"; }) .exceptionally( e -> { log.error("exceptionally exception",e); return "出異常了。。"; }); log.info("ret ====> {}", f.get()); // 執行結果：ret ====> 出異常了。。 // 假如不拋出異常，執行結果：ret ====> Hello World!- 無關性任務，互相依賴，
allOf
CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> "hello"); CompletableFuture<String> f4 = CompletableFuture.supplyAsync(() -> "world"); CompletableFuture<String> f5 = CompletableFuture.supplyAsync( () -> { try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } return "!"; }); // 使用allOf方法 f3 f4 f5 都執行結束之前一直阻塞 CompletableFuture.allOf(f3, f4, f5).join(); System.out.println(f3.get()); System.out.println(f4.get()); System.out.println(f5.get()); List<String> r = Stream.of(f3, f4, f5).map(CompletableFuture::join).collect(Collectors.toList()); System.out.println(r); // 執行結果：hello // world // ! // [hello, world, !] // 而且要等f1，f2,f3 三個任務都結束，不然會一直阻塞。這個類中的大部分方法上面都做了介紹，下面可以結合具體場景做一次演示。
- 異步執行任務，不返回結果：
三：DEMO
場景1：需要查詢一個訂單信息，首先需要查詢商品信息，然後查詢支付信息，最後匯總成一個對象返回。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "商品信息")
.thenCombineAsync(CompletableFuture.supplyAsync(() -> "支付信息"), (a, b) -> {
// 組裝信息
return a + b;
});
log.info("ret =========>{}",future.get());
場景2：用戶註冊，首先需要校驗用戶信息，然後生成賬號信息，最後保存到數據庫。這三個操作互相依賴。
// A -> B-> C
CompletableFuture<String> future = CompletableFuture.runAsync(() -> {
if (ThreadLocalRandom.current().nextBoolean()){
return;
}
throw new RuntimeException("該手機號碼已經註冊");
}).thenCompose(ret -> CompletableFuture.supplyAsync(() -> {
if (ThreadLocalRandom.current().nextBoolean()) {
// 生成賬號信息
return "賬號信息: 16289";
}
throw new RuntimeException("賬號信息生成失敗。。");
})).thenApplyAsync(ret -> {
// 保存賬號信息
log.info("保存賬號信息->{}", ret);
return "註冊成功";
}).exceptionally(e -> "註冊失敗" + e.getMessage());
log.info("最終返回結果：===》 {}",future.get());
