成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

Java 8中的CompletableFuture太好用了!20個示例全分享…

開發 后端
這篇文章介紹 Java 8 的 CompletionStage API 和它的標準庫的實現 CompletableFuture。API通過例子的方式演示了它的行為,每個例子演示一到兩個行為。

[[388339]]

在Java中異步編程,不一定非要使用rxJava, Java本身的庫中的CompletableFuture可以很好的應對大部分的場景。

這篇文章介紹 Java 8 的 CompletionStage API 和它的標準庫的實現 CompletableFuture。API通過例子的方式演示了它的行為,每個例子演示一到兩個行為。

既然CompletableFuture類實現了CompletionStage接口,首先我們需要理解這個接口的契約。它代表了一個特定的計算的階段,可以同步或者異步的被完成。你可以把它看成一個計算流水線上的一個單元,最終會產生一個最終結果,這意味著幾個CompletionStage可以串聯起來,一個完成的階段可以觸發下一階段的執行,接著觸發下一次,接著……

除了實現CompletionStage接口, CompletableFuture也實現了future接口, 代表一個未完成的異步事件。CompletableFuture提供了方法,能夠顯式地完成這個future,所以它叫CompletableFuture。

1、創建一個完成的CompletableFuture

最簡單的例子就是使用一個預定義的結果創建一個完成的CompletableFuture,通常我們會在計算的開始階段使用它。 

  1. static void completedFutureExample() {  
  2.     CompletableFuture cf = CompletableFuture.completedFuture("message");  
  3.     assertTrue(cf.isDone());  
  4.     assertEquals("message", cf.getNow(null));  

getNow(null)方法在future完成的情況下會返回結果,就比如上面這個例子,否則返回null (傳入的參數)。

2、運行一個簡單的異步階段

這個例子創建一個一個異步執行的階段: 

  1. static void runAsyncExample() {  
  2.     CompletableFuture cf = CompletableFuture.runAsync(() -> {  
  3.         assertTrue(Thread.currentThread().isDaemon());  
  4.         randomSleep();  
  5.     });  
  6.     assertFalse(cf.isDone());  
  7.     sleepEnough();  
  8.     assertTrue(cf.isDone());  

通過這個例子可以學到兩件事情:

CompletableFuture的方法如果以Async結尾,它會異步的執行(沒有指定executor的情況下), 異步執行通過ForkJoinPool實現, 它使用守護線程去執行任務。注意這是CompletableFuture的特性, 其它CompletionStage可以override這個默認的行為。

參考閱讀:任務并行執行神器:Fork&Join框架

3、在前一個階段上應用函數

下面這個例子使用前面 #1 的完成的CompletableFuture, #1返回結果為字符串message,然后應用一個函數把它變成大寫字母。 

  1. static void thenApplyExample() {  
  2.     CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> {  
  3.         assertFalse(Thread.currentThread().isDaemon());  
  4.         return s.toUpperCase();  
  5.     });  
  6.     assertEquals("MESSAGE", cf.getNow(null));  

注意thenApply方法名稱代表的行為。

then意味著這個階段的動作發生當前的階段正常完成之后。本例中,當前節點完成,返回字符串message。

Apply意味著返回的階段將會對結果前一階段的結果應用一個函數。

函數的執行會被阻塞,這意味著getNow()只有打斜操作被完成后才返回。

另外,關注公眾號Java技術棧,在后臺回復:面試,可以獲取我整理的 Java 并發多線程系列面試題和答案,非常齊全。

4、在前一個階段上異步應用函數

通過調用異步方法(方法后邊加Async后綴),串聯起來的CompletableFuture可以異步地執行(使用ForkJoinPool.commonPool())。 

  1. static void thenApplyAsyncExample() {  
  2.     CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {  
  3.         assertTrue(Thread.currentThread().isDaemon());  
  4.         randomSleep();  
  5.         return s.toUpperCase();  
  6.     });  
  7.     assertNull(cf.getNow(null));  
  8.     assertEquals("MESSAGE", cf.join());  

5、使用定制的Executor在前一個階段上異步應用函數

異步方法一個非常有用的特性就是能夠提供一個Executor來異步地執行CompletableFuture?!毒€程池全面解析》推薦看下。

這個例子演示了如何使用一個固定大小的線程池來應用大寫函數。 

  1. static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() {  
  2.     int count = 1;   
  3.     @Override  
  4.     public Thread newThread(Runnable runnable) {  
  5.         return new Thread(runnable, "custom-executor-" + count++);  
  6.     }  
  7. });  
  8. static void thenApplyAsyncWithExecutorExample() {  
  9.     CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {  
  10.         assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));  
  11.         assertFalse(Thread.currentThread().isDaemon());  
  12.         randomSleep();  
  13.         return s.toUpperCase();  
  14.     }, executor);  
  15.     assertNull(cf.getNow(null));  
  16.     assertEquals("MESSAGE", cf.join());  

6、消費前一階段的結果

如果下一階段接收了當前階段的結果,但是在計算的時候不需要返回值(它的返回類型是void), 那么它可以不應用一個函數,而是一個消費者, 調用方法也變成了thenAccept: 

  1. static void thenAcceptExample() {  
  2.     StringBuilder result = new StringBuilder();  
  3.     CompletableFuture.completedFuture("thenAccept message")  
  4.             .thenAccept(s -> result.append(s));  
  5.     assertTrue("Result was empty", result.length() > 0);  

本例中消費者同步地執行,所以我們不需要在CompletableFuture調用join方法。 

7、異步地消費遷移階段的結果

同樣,可以使用thenAcceptAsync方法, 串聯的CompletableFuture可以異步地執行。 

  1. static void thenAcceptAsyncExample() {  
  2.     StringBuilder result = new StringBuilder();  
  3.     CompletableFuture cf = CompletableFuture.completedFuture("thenAcceptAsync message")  
  4.             .thenAcceptAsync(s -> result.append(s));  
  5.     cf.join();  
  6.     assertTrue("Result was empty", result.length() > 0);  

8、完成計算異常

現在我們來看一下異步操作如何顯式地返回異常,用來指示計算失敗。我們簡化這個例子,操作處理一個字符串,把它轉換成答謝,我們模擬延遲一秒。

我們使用thenApplyAsync(Function, Executor)方法,第一個參數傳入大寫函數, executor是一個delayed executor,在執行前會延遲一秒。 

  1. static void completeExceptionallyExample() {  
  2.     CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,  
  3.             CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));  
  4.     CompletableFuture exceptionHandler = cf.handle((s, th) -> { return (th != null) ? "message upon cancel" : ""; });  
  5.     cf.completeExceptionally(new RuntimeException("completed exceptionally"));  
  6. assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());  
  7.     try {  
  8.         cf.join();  
  9.         fail("Should have thrown an exception");  
  10.     } catch(CompletionException ex) { // just for testing  
  11.         assertEquals("completed exceptionally", ex.getCause().getMessage());  
  12.     }  
  13.     assertEquals("message upon cancel", exceptionHandler.join());  

讓我們看一下細節。

首先我們創建了一個CompletableFuture, 完成后返回一個字符串message,接著我們調用thenApplyAsync方法,它返回一個CompletableFuture。這個方法在第一個函數完成后,異步地應用轉大寫字母函數。

這個例子還演示了如何通過delayedExecutor(timeout, timeUnit)延遲執行一個異步任務。

我們創建了一個分離的handler階段:exceptionHandler, 它處理異常異常,在異常情況下返回message upon cancel。

下一步我們顯式地用異常完成第二個階段。在階段上調用join方法,它會執行大寫轉換,然后拋出CompletionException(正常的join會等待1秒,然后得到大寫的字符串。不過我們的例子還沒等它執行就完成了異常), 然后它觸發了handler階段。

9、取消計算

和完成異常類似,我們可以調用cancel(boolean mayInterruptIfRunning)取消計算。對于CompletableFuture類,布爾參數并沒有被使用,這是因為它并沒有使用中斷去取消操作,相反,cancel等價于completeExceptionally(new CancellationException())。 

  1. static void cancelExample() {  
  2.     CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,  
  3.             CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));  
  4.     CompletableFuture cfcf2 = cf.exceptionally(throwable -> "canceled message");  
  5.     assertTrue("Was not canceled", cf.cancel(true));  
  6.     assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());  
  7.     assertEquals("canceled message", cf2.join());  

10、在兩個完成的階段其中之一上應用函數

下面的例子創建了CompletableFuture, applyToEither處理兩個階段, 在其中之一上應用函數(包保證哪一個被執行)。本例中的兩個階段一個是應用大寫轉換在原始的字符串上, 另一個階段是應用小些轉換。 

  1. static void applyToEitherExample() {  
  2.     String original = "Message" 
  3.     CompletableFuture cf1 = CompletableFuture.completedFuture(original)  
  4.             .thenApplyAsync(s -> delayedUpperCase(s));  
  5.     CompletableFuture cf2 = cf1.applyToEither(  
  6.             CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),  
  7.             s -> s + " from applyToEither");  
  8.     assertTrue(cf2.join().endsWith(" from applyToEither"));  

11、在兩個完成的階段其中之一上調用消費函數

和前一個例子很類似了,只不過我們調用的是消費者函數 (Function變成Consumer): 

  1. static void acceptEitherExample() {  
  2.     String original = "Message" 
  3.     StringBuilder result = new StringBuilder();  
  4.     CompletableFuture cf = CompletableFuture.completedFuture(original)  
  5.             .thenApplyAsync(s -> delayedUpperCase(s))  
  6.             .acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),  
  7.                     s -> result.append(s).append("acceptEither"));  
  8.     cf.join();  
  9.     assertTrue("Result was empty", result.toString().endsWith("acceptEither"));  

12、在兩個階段都執行完后運行一個 Runnable

這個例子演示了依賴的CompletableFuture如果等待兩個階段完成后執行了一個Runnable。 

注意下面所有的階段都是同步執行的,第一個階段執行大寫轉換,第二個階段執行小寫轉換。 

  1. static void runAfterBothExample() {  
  2.     String original = "Message" 
  3.     StringBuilder result = new StringBuilder();  
  4.     CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth(  
  5.             CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),  
  6.             () -> result.append("done"));  
  7.     assertTrue("Result was empty", result.length() > 0);  

13、 使用BiConsumer處理兩個階段的結果

上面的例子還可以通過BiConsumer來實現: 

  1. static void thenAcceptBothExample() {  
  2.     String original = "Message" 
  3.     StringBuilder result = new StringBuilder();  
  4.     CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth(  
  5.             CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),  
  6.             (s1, s2) -> result.append(s1 + s2));  
  7.     assertEquals("MESSAGEmessage", result.toString());  

14、使用BiFunction處理兩個階段的結果

如果CompletableFuture依賴兩個前面階段的結果, 它復合兩個階段的結果再返回一個結果,我們就可以使用thenCombine()函數。整個流水線是同步的,所以getNow()會得到最終的結果,它把大寫和小寫字符串連接起來。 

  1. static void thenCombineExample() {  
  2.     String original = "Message" 
  3.     CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))  
  4.             .thenCombine(CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)),  
  5.                     (s1, s2) -> s1 + s2);  
  6.     assertEquals("MESSAGEmessage", cf.getNow(null));  

15、異步使用BiFunction處理兩個階段的結果

類似上面的例子,但是有一點不同:依賴的前兩個階段異步地執行,所以thenCombine()也異步地執行,即時它沒有Async后綴。

Javadoc中有注釋:

Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method

所以我們需要join方法等待結果的完成。 

  1. static void thenCombineAsyncExample() {  
  2.     String original = "Message" 
  3.     CompletableFuture cf = CompletableFuture.completedFuture(original)  
  4.             .thenApplyAsync(s -> delayedUpperCase(s))  
  5.             .thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),  
  6.                     (s1, s2) -> s1 + s2);  
  7.     assertEquals("MESSAGEmessage", cf.join());  

16、組合 CompletableFuture

我們可以使用thenCompose()完成上面兩個例子。這個方法等待第一個階段的完成(大寫轉換), 它的結果傳給一個指定的返回CompletableFuture函數,它的結果就是返回的CompletableFuture的結果。

有點拗口,但是我們看例子來理解。函數需要一個大寫字符串做參數,然后返回一個CompletableFuture, 這個CompletableFuture會轉換字符串變成小寫然后連接在大寫字符串的后面。 

  1. static void thenComposeExample() {  
  2.     String original = "Message" 
  3.     CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))  
  4.             .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))  
  5.                     .thenApply(s -> upper + s));  
  6.     assertEquals("MESSAGEmessage", cf.join());  

17、當幾個階段中的一個完成,創建一個完成的階段

下面的例子演示了當任意一個CompletableFuture完成后, 創建一個完成的CompletableFuture.

待處理的階段首先創建, 每個階段都是轉換一個字符串為大寫。因為本例中這些階段都是同步地執行(thenApply), 從anyOf中創建的CompletableFuture會立即完成,這樣所有的階段都已完成,我們使用whenComplete(BiConsumer<? super Object, ? super Throwable> action)處理完成的結果。 

  1. static void anyOfExample() {  
  2.     StringBuilder result = new StringBuilder();  
  3.     List messages = Arrays.asList("a", "b", "c");  
  4.     List<CompletableFuture> futures = messages.stream()  
  5.             .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))  
  6.             .collect(Collectors.toList());  
  7.     CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((res, th) -> {  
  8.         if(th == null) {  
  9.             assertTrue(isUpperCase((String) res));  
  10.             result.append(res);  
  11.         }  
  12.     });  
  13.     assertTrue("Result was empty", result.length() > 0);  

18、當所有的階段都完成后創建一個階段

上一個例子是當任意一個階段完成后接著處理,接下來的兩個例子演示當所有的階段完成后才繼續處理, 同步地方式和異步地方式兩種。 

  1. static void allOfExample() {  
  2.     StringBuilder result = new StringBuilder();  
  3.     List messages = Arrays.asList("a", "b", "c");  
  4.     List<CompletableFuture> futures = messages.stream()  
  5.             .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))  
  6.             .collect(Collectors.toList());  
  7.     CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((v, th) -> {  
  8.         futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));  
  9.         result.append("done");  
  10.     });  
  11.     assertTrue("Result was empty", result.length() > 0);  

19、當所有的階段都完成后異步地創建一個階段

使用thenApplyAsync()替換那些單個的CompletableFutures的方法,allOf()會在通用池中的線程中異步地執行。所以我們需要調用join方法等待它完成。 

  1. static void allOfAsyncExample() {  
  2.     StringBuilder result = new StringBuilder();  
  3.     List messages = Arrays.asList("a", "b", "c");  
  4.     List<CompletableFuture> futures = messages.stream()  
  5.             .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> delayedUpperCase(s)))  
  6.             .collect(Collectors.toList());  
  7.     CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))  
  8.             .whenComplete((v, th) -> { 
  9.                 futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));  
  10.                 result.append("done");  
  11.             });  
  12.     allOf.join();  
  13.     assertTrue("Result was empty", result.length() > 0);  

20、真實的例子

現在你已經了解了CompletionStage 和 CompletableFuture 的一些函數的功能,下面的例子是一個實踐場景:

  1.  首先異步調用cars方法獲得Car的列表,它返回CompletionStage場景。cars消費一個遠程的REST API。
  2.  然后我們復合一個CompletionStage填寫每個汽車的評分,通過rating(manufacturerId)返回一個CompletionStage, 它會異步地獲取汽車的評分(可能又是一個REST API調用)
  3.  當所有的汽車填好評分后,我們結束這個列表,所以我們調用allOf得到最終的階段, 它在前面階段所有階段完成后才完成。
  4.  在最終的階段調用whenComplete(),我們打印出每個汽車和它的評分。 
  1. cars().thenCompose(cars -> {  
  2.     List<CompletionStage> updatedCars = cars.stream()  
  3.             .map(car -> rating(car.manufacturerId).thenApply(r -> {  
  4.                 car.setRating(r);  
  5.                 return car;  
  6.             })).collect(Collectors.toList());  
  7.     CompletableFuture done = CompletableFuture  
  8.             .allOf(updatedCars.toArray(new CompletableFuture[updatedCars.size()]));  
  9.     return done.thenApply(v -> updatedCars.stream().map(CompletionStage::toCompletableFuture)  
  10.             .map(CompletableFuture::join).collect(Collectors.toList()));  
  11. }).whenComplete((cars, th) -> {  
  12.     if (th == null) { 
  13.         cars.forEach(System.out::println);  
  14.     } else {  
  15.         throw new RuntimeException(th);  
  16.     }  
  17. }).toCompletableFuture().join(); 

因為每個汽車的實例都是獨立的,得到每個汽車的評分都可以異步地執行,這會提高系統的性能(延遲),而且,等待所有的汽車評分被處理使用的是allOf方法,而不是手工的線程等待(Thread#join() 或 a CountDownLatch)。

 

責任編輯:龐桂玉 來源: Java技術棧
相關推薦

2021-08-11 09:33:15

Vue 技巧 開發工具

2024-12-13 16:01:35

2021-02-21 14:35:29

Java 8異步編程

2022-07-14 08:36:28

NacosApollo長輪詢

2021-04-22 09:56:32

MYSQL開發數據庫

2022-08-01 07:02:06

SpringEasyExcel場景

2024-05-11 09:38:05

React編譯器React 19

2022-05-31 09:42:49

工具編輯器

2021-03-19 09:48:10

Jupyter Not插件Python

2020-06-23 15:58:42

心電圖

2020-12-29 10:45:55

開發設計代碼

2022-09-06 10:52:04

正則庫HumrePython

2016-11-29 12:46:24

JavaJava8時間日期庫

2024-04-18 08:20:27

Java 8編程工具

2021-09-10 10:15:24

Python人臉識別AI

2022-05-11 14:43:37

WindowsPython服務器

2022-07-25 06:42:24

分布式鎖Redis

2020-11-10 06:11:59

工具軟件代碼

2021-03-02 20:42:20

實戰策略

2022-06-28 07:14:23

WizTree磁盤文件清理
點贊
收藏

51CTO技術棧公眾號

主站蜘蛛池模板: 日韩欧美中文字幕在线视频 | 亚洲精品电影在线 | 国产成人精品一区二 | 九一国产精品 | 桃花av在线| 欧美jizzhd精品欧美巨大免费 | 亚州成人 | 日韩在线不卡视频 | 国产成人免费网站 | 免费骚视频 | 久久国产精品一区二区三区 | 亚洲欧美中文日韩在线v日本 | 欧美激情一区二区三区 | 久久噜噜噜精品国产亚洲综合 | 精品久久久久久久久久 | 一区二区三区亚洲视频 | 欧美bondage紧缚视频 | 久久久久国| 亚洲电影专区 | 丝袜 亚洲 欧美 日韩 综合 | 欧美日韩一区二区在线 | 国产激情福利 | av入口 | 草久视频 | 99综合在线 | 欧美一区二区免费电影 | 欧美黑人狂野猛交老妇 | 宅男伊人| 青娱乐国产 | 国产在线精品一区二区 | 午夜视频在线观看网站 | 精品日韩在线 | 久草精品视频 | 国产伦精品一区二区三区照片91 | 欧美精品一区在线发布 | 国产真实精品久久二三区 | 国产激情自拍视频 | 日韩中文一区二区三区 | 欧美一区二区三区 | 亚洲精品一区二三区不卡 | 欧美一级黄色免费看 |