偷偷摘套内射激情视频,久久精品99国产国产精,中文字幕无线乱码人妻,中文在线中文a,性爽19p

異步編程利器：CompletableFuture詳解

作者：撿田螺的小男孩
開(kāi)發(fā) 項(xiàng)目管理
最近剛好使用CompeletableFuture優(yōu)化了項(xiàng)目中的代碼，所以跟大家一起學(xué)習(xí)CompletableFuture。

[[403915]]

前言

最近剛好使用CompeletableFuture優(yōu)化了項(xiàng)目中的代碼，所以跟大家一起學(xué)習(xí)CompletableFuture。

一個(gè)例子回顧 Future

因?yàn)镃ompletableFuture實(shí)現(xiàn)了Future接口，我們先來(lái)回顧Future吧。

Future是Java5新加的一個(gè)接口，它提供了一種異步并行計(jì)算的功能。如果主線程需要執(zhí)行一個(gè)很耗時(shí)的計(jì)算任務(wù)，我們就可以通過(guò)future把這個(gè)任務(wù)放到異步線程中執(zhí)行。主線程繼續(xù)處理其他任務(wù)，處理完成后，再通過(guò)Future獲取計(jì)算結(jié)果。

來(lái)看個(gè)簡(jiǎn)單例子吧，假設(shè)我們有兩個(gè)任務(wù)服務(wù)，一個(gè)查詢用戶基本信息，一個(gè)是查詢用戶勛章信息。如下，

  1. public class UserInfoService { 
  2.  
  3.     public UserInfo getUserInfo(Long userId) throws InterruptedException { 
  4.         Thread.sleep(300);//模擬調(diào)用耗時(shí) 
  5.         return new UserInfo("666", "撿田螺的小男孩", 27); //一般是查數(shù)據(jù)庫(kù)，或者遠(yuǎn)程調(diào)用返回的 
  6.     } 
  7. } 
  8.  
  9. public class MedalService { 
  10.  
  11.     public MedalInfo getMedalInfo(long userId) throws InterruptedException { 
  12.         Thread.sleep(500); //模擬調(diào)用耗時(shí) 
  13.         return new MedalInfo("666", "守護(hù)勛章"); 
  14.     } 
  15. } 

接下來(lái)，我們來(lái)演示下，在主線程中是如何使用Future來(lái)進(jìn)行異步調(diào)用的。

  1. public class FutureTest { 
  2.  
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  4.  
  5.         ExecutorService executorService = Executors.newFixedThreadPool(10); 
  6.  
  7.         UserInfoService userInfoService = new UserInfoService(); 
  8.         MedalService medalService = new MedalService(); 
  9.         long userId =666L; 
  10.         long startTime = System.currentTimeMillis(); 
  11.  
  12.         //調(diào)用用戶服務(wù)獲取用戶基本信息 
  13.         FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() { 
  14.             @Override 
  15.             public UserInfo call() throws Exception { 
  16.                 return userInfoService.getUserInfo(userId); 
  17.             } 
  18.         }); 
  19.         executorService.submit(userInfoFutureTask); 
  20.  
  21.         Thread.sleep(300); //模擬主線程其它操作耗時(shí) 
  22.  
  23.         FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() { 
  24.             @Override 
  25.             public MedalInfo call() throws Exception { 
  26.                 return medalService.getMedalInfo(userId); 
  27.             } 
  28.         }); 
  29.         executorService.submit(medalInfoFutureTask); 
  30.  
  31.         UserInfo userInfo = userInfoFutureTask.get();//獲取個(gè)人信息結(jié)果 
  32.         MedalInfo medalInfo = medalInfoFutureTask.get();//獲取勛章信息結(jié)果 
  33.  
  34.         System.out.println("總共用時(shí)" + (System.currentTimeMillis() - startTime) + "ms"); 
  35.     } 
  36. } 
  37.      

運(yùn)行結(jié)果:

  1. 總共用時(shí)806ms 

如果我們不使用Future進(jìn)行并行異步調(diào)用，而是在主線程串行進(jìn)行的話，耗時(shí)大約為300+500+300 = 1100 ms?？梢园l(fā)現(xiàn)，future+線程池異步配合，提高了程序的執(zhí)行效率。

但是Future對(duì)于結(jié)果的獲取，不是很友好，只能通過(guò)阻塞或者輪詢的方式得到任務(wù)的結(jié)果。

  • Future.get() 就是阻塞調(diào)用，在線程獲取結(jié)果之前get方法會(huì)一直阻塞。
  • Future提供了一個(gè)isDone方法，可以在程序中輪詢這個(gè)方法查詢執(zhí)行結(jié)果。

阻塞的方式和異步編程的設(shè)計(jì)理念相違背，而輪詢的方式會(huì)耗費(fèi)無(wú)謂的CPU資源。因此，JDK8設(shè)計(jì)出CompletableFuture。CompletableFuture提供了一種觀察者模式類似的機(jī)制，可以讓任務(wù)執(zhí)行完成后通知監(jiān)聽(tīng)的一方。

一個(gè)例子走進(jìn)CompletableFuture

我們還是基于以上Future的例子，改用CompletableFuture 來(lái)實(shí)現(xiàn)

  1. public class FutureTest { 
  2.  
  3.     public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { 
  4.  
  5.         UserInfoService userInfoService = new UserInfoService(); 
  6.         MedalService medalService = new MedalService(); 
  7.         long userId =666L; 
  8.         long startTime = System.currentTimeMillis(); 
  9.  
  10.         //調(diào)用用戶服務(wù)獲取用戶基本信息 
  11.         CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId)); 
  12.  
  13.         Thread.sleep(300); //模擬主線程其它操作耗時(shí) 
  14.  
  15.         CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId));  
  16.  
  17.         UserInfo userInfo = completableUserInfoFuture.get(2,TimeUnit.SECONDS);//獲取個(gè)人信息結(jié)果 
  18.         MedalInfo medalInfo = completableMedalInfoFuture.get();//獲取勛章信息結(jié)果 
  19.         System.out.println("總共用時(shí)" + (System.currentTimeMillis() - startTime) + "ms"); 
  20.  
  21.     } 
  22. } 

可以發(fā)現(xiàn)，使用CompletableFuture，代碼簡(jiǎn)潔了很多。CompletableFuture的supplyAsync方法，提供了異步執(zhí)行的功能，線程池也不用單獨(dú)創(chuàng)建了。實(shí)際上，它CompletableFuture使用了默認(rèn)線程池是ForkJoinPool.commonPool。

CompletableFuture提供了幾十種方法，輔助我們的異步任務(wù)場(chǎng)景。這些方法包括創(chuàng)建異步任務(wù)、任務(wù)異步回調(diào)、多個(gè)任務(wù)組合處理等方面。我們一起來(lái)學(xué)習(xí)吧

CompletableFuture使用場(chǎng)景

創(chuàng)建異步任務(wù)

CompletableFuture創(chuàng)建異步任務(wù)，一般有supplyAsync和runAsync兩個(gè)方法

創(chuàng)建異步任務(wù)

  • supplyAsync執(zhí)行CompletableFuture任務(wù)，支持返回值
  • runAsync執(zhí)行CompletableFuture任務(wù)，沒(méi)有返回值。

supplyAsync方法

//使用默認(rèn)內(nèi)置線程池ForkJoinPool.commonPool()，根據(jù)supplier構(gòu)建執(zhí)行任務(wù)

  1. //使用默認(rèn)內(nèi)置線程池ForkJoinPool.commonPool()，根據(jù)supplier構(gòu)建執(zhí)行任務(wù) 
  2. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) 
  3. //自定義線程，根據(jù)supplier構(gòu)建執(zhí)行任務(wù) 
  4. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) 

runAsync方法

  1. //使用默認(rèn)內(nèi)置線程池ForkJoinPool.commonPool()，根據(jù)runnable構(gòu)建執(zhí)行任務(wù) 
  2. public static CompletableFuture<Void> runAsync(Runnable runnable)  
  3. //自定義線程，根據(jù)runnable構(gòu)建執(zhí)行任務(wù) 
  4. public static CompletableFuture<Void> runAsync(Runnable runnable,  Executor executor) 

實(shí)例代碼如下：

  1. public class FutureTest { 
  2.  
  3.     public static void main(String[] args) { 
  4.         //可以自定義線程池 
  5.         ExecutorService executor = Executors.newCachedThreadPool(); 
  6.         //runAsync的使用 
  7.         CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run,關(guān)注公眾號(hào):撿田螺的小男孩"), executor); 
  8.         //supplyAsync的使用 
  9.         CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> { 
  10.                     System.out.print("supply,關(guān)注公眾號(hào):撿田螺的小男孩"); 
  11.                     return "撿田螺的小男孩"; }, executor); 
  12.         //runAsync的future沒(méi)有返回值，輸出null 
  13.         System.out.println(runFuture.join()); 
  14.         //supplyAsync的future，有返回值 
  15.         System.out.println(supplyFuture.join()); 
  16.         executor.shutdown(); // 線程池需要關(guān)閉 
  17.     } 
  18. } 
  19. //輸出 
  20. run,關(guān)注公眾號(hào):撿田螺的小男孩 
  21. null 
  22. supply,關(guān)注公眾號(hào):撿田螺的小男孩撿田螺的小男孩 

任務(wù)異步回調(diào)

1. thenRun/thenRunAsync

  1. public CompletableFuture<Void> thenRun(Runnable action); 
  2. public CompletableFuture<Void> thenRunAsync(Runnable action); 

CompletableFuture的thenRun方法，通俗點(diǎn)講就是，做完第一個(gè)任務(wù)后，再做第二個(gè)任務(wù)。某個(gè)任務(wù)執(zhí)行完成后，執(zhí)行回調(diào)方法;但是前后兩個(gè)任務(wù)沒(méi)有參數(shù)傳遞，第二個(gè)任務(wù)也沒(méi)有返回值

  1. public class FutureThenRunTest { 
  2.  
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  4.  
  5.         CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync( 
  6.                 ()->{ 
  7.                     System.out.println("先執(zhí)行第一個(gè)CompletableFuture方法任務(wù)"); 
  8.                     return "撿田螺的小男孩"; 
  9.                 } 
  10.         ); 
  11.  
  12.         CompletableFuture thenRunFuture = orgFuture.thenRun(() -> { 
  13.             System.out.println("接著執(zhí)行第二個(gè)任務(wù)"); 
  14.         }); 
  15.  
  16.         System.out.println(thenRunFuture.get()); 
  17.     } 
  18. } 
  19. //輸出 
  20. 先執(zhí)行第一個(gè)CompletableFuture方法任務(wù) 
  21. 接著執(zhí)行第二個(gè)任務(wù) 
  22. null 

thenRun 和thenRunAsync有什么區(qū)別呢?可以看下源碼哈：

  1. private static final Executor asyncPool = useCommonPool ? 
  2.       ForkJoinPool.commonPool() : new ThreadPerTaskExecutor(); 
  3.        
  4.   public CompletableFuture<Void> thenRun(Runnable action) { 
  5.       return uniRunStage(null, action); 
  6.   } 
  7.  
  8.   public CompletableFuture<Void> thenRunAsync(Runnable action) { 
  9.       return uniRunStage(asyncPool, action); 
  10.   } 

如果你執(zhí)行第一個(gè)任務(wù)的時(shí)候，傳入了一個(gè)自定義線程池：

  • 調(diào)用thenRun方法執(zhí)行第二個(gè)任務(wù)時(shí)，則第二個(gè)任務(wù)和第一個(gè)任務(wù)是共用同一個(gè)線程池。
  • 調(diào)用thenRunAsync執(zhí)行第二個(gè)任務(wù)時(shí)，則第一個(gè)任務(wù)使用的是你自己傳入的線程池，第二個(gè)任務(wù)使用的是ForkJoin線程池

TIPS: 后面介紹的thenAccept和thenAcceptAsync，thenApply和thenApplyAsync等，它們之間的區(qū)別也是這個(gè)哈。

2.thenAccept/thenAcceptAsync

CompletableFuture的thenAccept方法表示，第一個(gè)任務(wù)執(zhí)行完成后，執(zhí)行第二個(gè)回調(diào)方法任務(wù)，會(huì)將該任務(wù)的執(zhí)行結(jié)果，作為入?yún)?，傳遞到回調(diào)方法中，但是回調(diào)方法是沒(méi)有返回值的。

  1. public class FutureThenAcceptTest { 
  2.  
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  4.  
  5.         CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync( 
  6.                 ()->{ 
  7.                     System.out.println("原始CompletableFuture方法任務(wù)"); 
  8.                     return "撿田螺的小男孩"; 
  9.                 } 
  10.         ); 
  11.  
  12.         CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> { 
  13.             if ("撿田螺的小男孩".equals(a)) { 
  14.                 System.out.println("關(guān)注了"); 
  15.             } 
  16.  
  17.             System.out.println("先考慮考慮"); 
  18.         }); 
  19.  
  20.         System.out.println(thenAcceptFuture.get()); 
  21.     } 
  22. } 

3. thenApply/thenApplyAsync

CompletableFuture的thenApply方法表示，第一個(gè)任務(wù)執(zhí)行完成后，執(zhí)行第二個(gè)回調(diào)方法任務(wù)，會(huì)將該任務(wù)的執(zhí)行結(jié)果，作為入?yún)?，傳遞到回調(diào)方法中，并且回調(diào)方法是有返回值的。

  1. public class FutureThenApplyTest { 
  2.  
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  4.  
  5.         CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync( 
  6.                 ()->{ 
  7.                     System.out.println("原始CompletableFuture方法任務(wù)"); 
  8.                     return "撿田螺的小男孩"; 
  9.                 } 
  10.         ); 
  11.  
  12.         CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> { 
  13.             if ("撿田螺的小男孩".equals(a)) { 
  14.                 return "關(guān)注了"; 
  15.             } 
  16.  
  17.             return "先考慮考慮"; 
  18.         }); 
  19.  
  20.         System.out.println(thenApplyFuture.get()); 
  21.     } 
  22. } 
  23. //輸出 
  24. 原始CompletableFuture方法任務(wù) 
  25. 關(guān)注了 

4. exceptionally

CompletableFuture的exceptionally方法表示，某個(gè)任務(wù)執(zhí)行異常時(shí)，執(zhí)行的回調(diào)方法;并且有拋出異常作為參數(shù)，傳遞到回調(diào)方法。

  1. public class FutureExceptionTest { 
  2.  
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  4.  
  5.         CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync( 
  6.                 ()->{ 
  7.                     System.out.println("當(dāng)前線程名稱：" + Thread.currentThread().getName()); 
  8.                     throw new RuntimeException(); 
  9.                 } 
  10.         ); 
  11.  
  12.         CompletableFuture<String> exceptionFuture = orgFuture.exceptionally((e) -> { 
  13.             e.printStackTrace(); 
  14.             return "你的程序異常啦"; 
  15.         }); 
  16.  
  17.         System.out.println(exceptionFuture.get()); 
  18.     } 
  19. } 
  20. //輸出 
  21. 當(dāng)前線程名稱：ForkJoinPool.commonPool-worker-1 
  22. java.util.concurrent.CompletionException: java.lang.RuntimeException 
  23.  at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) 
  24.  at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280) 
  25.  at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592) 
  26.  at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582) 
  27.  at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) 
  28.  at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) 
  29.  at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) 
  30.  at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157) 
  31. Caused by: java.lang.RuntimeException 
  32.  at cn.eovie.future.FutureWhenTest.lambda$main$0(FutureWhenTest.java:13) 
  33.  at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) 
  34.  ... 5 more 
  35. 你的程序異常啦 

5. whenComplete方法

CompletableFuture的whenComplete方法表示，某個(gè)任務(wù)執(zhí)行完成后，執(zhí)行的回調(diào)方法，無(wú)返回值;并且whenComplete方法返回的CompletableFuture的result是上個(gè)任務(wù)的結(jié)果。

  1. public class FutureWhenTest { 
  2.  
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  4.  
  5.         CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync( 
  6.                 ()->{ 
  7.                     System.out.println("當(dāng)前線程名稱：" + Thread.currentThread().getName()); 
  8.                     try { 
  9.                         Thread.sleep(2000L); 
  10.                     } catch (InterruptedException e) { 
  11.                         e.printStackTrace(); 
  12.                     } 
  13.                     return "撿田螺的小男孩"; 
  14.                 } 
  15.         ); 
  16.  
  17.         CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> { 
  18.             System.out.println("當(dāng)前線程名稱：" + Thread.currentThread().getName()); 
  19.             System.out.println("上個(gè)任務(wù)執(zhí)行完啦，還把" + a + "傳過(guò)來(lái)"); 
  20.             if ("撿田螺的小男孩".equals(a)) { 
  21.                 System.out.println("666"); 
  22.             } 
  23.             System.out.println("233333"); 
  24.         }); 
  25.  
  26.         System.out.println(rstFuture.get()); 
  27.     } 
  28. } 
  29. //輸出 
  30. 當(dāng)前線程名稱：ForkJoinPool.commonPool-worker-1 
  31. 當(dāng)前線程名稱：ForkJoinPool.commonPool-worker-1 
  32. 上個(gè)任務(wù)執(zhí)行完啦，還把撿田螺的小男孩傳過(guò)來(lái) 
  33. 666 
  34. 233333 
  35. 撿田螺的小男孩 

6. handle方法

CompletableFuture的handle方法表示，某個(gè)任務(wù)執(zhí)行完成后，執(zhí)行回調(diào)方法，并且是有返回值的;并且handle方法返回的CompletableFuture的result是回調(diào)方法執(zhí)行的結(jié)果。

  1. public class FutureHandlerTest { 
  2.  
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  4.  
  5.         CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync( 
  6.                 ()->{ 
  7.                     System.out.println("當(dāng)前線程名稱：" + Thread.currentThread().getName()); 
  8.                     try { 
  9.                         Thread.sleep(2000L); 
  10.                     } catch (InterruptedException e) { 
  11.                         e.printStackTrace(); 
  12.                     } 
  13.                     return "撿田螺的小男孩"; 
  14.                 } 
  15.         ); 
  16.  
  17.         CompletableFuture<String> rstFuture = orgFuture.handle((a, throwable) -> { 
  18.  
  19.             System.out.println("上個(gè)任務(wù)執(zhí)行完啦，還把" + a + "傳過(guò)來(lái)"); 
  20.             if ("撿田螺的小男孩".equals(a)) { 
  21.                 System.out.println("666"); 
  22.                 return "關(guān)注了"; 
  23.             } 
  24.             System.out.println("233333"); 
  25.             return null; 
  26.         }); 
  27.  
  28.         System.out.println(rstFuture.get()); 
  29.     } 
  30. } 
  31. //輸出 
  32. 當(dāng)前線程名稱：ForkJoinPool.commonPool-worker-1 
  33. 上個(gè)任務(wù)執(zhí)行完啦，還把撿田螺的小男孩傳過(guò)來(lái) 
  34. 666 
  35. 關(guān)注了 

多個(gè)任務(wù)組合處理

AND組合關(guān)系

thenCombine / thenAcceptBoth / runAfterBoth都表示：將兩個(gè)CompletableFuture組合起來(lái)，只有這兩個(gè)都正常執(zhí)行完了，才會(huì)執(zhí)行某個(gè)任務(wù)。

區(qū)別在于：

  • thenCombine：會(huì)將兩個(gè)任務(wù)的執(zhí)行結(jié)果作為方法入?yún)?，傳遞到指定方法中，且有返回值
  • thenAcceptBoth: 會(huì)將兩個(gè)任務(wù)的執(zhí)行結(jié)果作為方法入?yún)?，傳遞到指定方法中，且無(wú)返回值
  • runAfterBoth 不會(huì)把執(zhí)行結(jié)果當(dāng)做方法入?yún)?，且沒(méi)有返回值。
  1. public class ThenCombineTest { 
  2.  
  3.     public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { 
  4.  
  5.         CompletableFuture<String> first = CompletableFuture.completedFuture("第一個(gè)異步任務(wù)"); 
  6.         ExecutorService executor = Executors.newFixedThreadPool(10); 
  7.         CompletableFuture<String> future = CompletableFuture 
  8.                 //第二個(gè)異步任務(wù) 
  9.                 .supplyAsync(() -> "第二個(gè)異步任務(wù)", executor) 
  10.                 // (w, s) -> System.out.println(s) 是第三個(gè)任務(wù) 
  11.                 .thenCombineAsync(first, (s, w) -> { 
  12.                     System.out.println(w); 
  13.                     System.out.println(s); 
  14.                     return "兩個(gè)異步任務(wù)的組合"; 
  15.                 }, executor); 
  16.         System.out.println(future.join()); 
  17.         executor.shutdown(); 
  18.  
  19.     } 
  20. } 
  21. //輸出 
  22. 第一個(gè)異步任務(wù) 
  23. 第二個(gè)異步任務(wù) 
  24. 兩個(gè)異步任務(wù)的組合 

OR 組合的關(guān)系

applyToEither / acceptEither / runAfterEither 都表示：將兩個(gè)CompletableFuture組合起來(lái)，只要其中一個(gè)執(zhí)行完了,就會(huì)執(zhí)行某個(gè)任務(wù)。

區(qū)別在于：

  • applyToEither：會(huì)將已經(jīng)執(zhí)行完成的任務(wù)，作為方法入?yún)?，傳遞到指定方法中，且有返回值
  • acceptEither: 會(huì)將已經(jīng)執(zhí)行完成的任務(wù)，作為方法入?yún)?，傳遞到指定方法中，且無(wú)返回值
  • runAfterEither：不會(huì)把執(zhí)行結(jié)果當(dāng)做方法入?yún)?，且沒(méi)有返回值。
  1. public class AcceptEitherTest { 
  2.     public static void main(String[] args) { 
  3.         //第一個(gè)異步任務(wù)，休眠2秒，保證它執(zhí)行晚點(diǎn) 
  4.         CompletableFuture<String> first = CompletableFuture.supplyAsync(()->{ 
  5.             try{ 
  6.  
  7.                 Thread.sleep(2000L); 
  8.                 System.out.println("執(zhí)行完第一個(gè)異步任務(wù)");} 
  9.                 catch (Exception e){ 
  10.                     return "第一個(gè)任務(wù)異常"; 
  11.                 } 
  12.             return "第一個(gè)異步任務(wù)"; 
  13.         }); 
  14.         ExecutorService executor = Executors.newSingleThreadExecutor(); 
  15.         CompletableFuture<Void> future = CompletableFuture 
  16.                 //第二個(gè)異步任務(wù) 
  17.                 .supplyAsync(() -> { 
  18.                             System.out.println("執(zhí)行完第二個(gè)任務(wù)"); 
  19.                             return "第二個(gè)任務(wù)";} 
  20.                 , executor) 
  21.                 //第三個(gè)任務(wù) 
  22.                 .acceptEitherAsync(first, System.out::println, executor); 
  23.  
  24.         executor.shutdown(); 
  25.     } 
  26. } 
  27. //輸出 
  28. 執(zhí)行完第二個(gè)任務(wù) 
  29. 第二個(gè)任務(wù) 

AllOf

所有任務(wù)都執(zhí)行完成后，才執(zhí)行 allOf返回的CompletableFuture。如果任意一個(gè)任務(wù)異常，allOf的CompletableFuture，執(zhí)行g(shù)et方法，會(huì)拋出異常

  1. public class allOfFutureTest { 
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  3.  
  4.         CompletableFuture<Void> a = CompletableFuture.runAsync(()->{ 
  5.             System.out.println("我執(zhí)行完了"); 
  6.         }); 
  7.         CompletableFuture<Void> b = CompletableFuture.runAsync(() -> { 
  8.             System.out.println("我也執(zhí)行完了"); 
  9.         }); 
  10.         CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b).whenComplete((m,k)->{ 
  11.             System.out.println("finish"); 
  12.         }); 
  13.     } 
  14. } 
  15. //輸出 
  16. 我執(zhí)行完了 
  17. 我也執(zhí)行完了 
  18. finish 

AnyOf

任意一個(gè)任務(wù)執(zhí)行完，就執(zhí)行anyOf返回的CompletableFuture。如果執(zhí)行的任務(wù)異常，anyOf的CompletableFuture，執(zhí)行g(shù)et方法，會(huì)拋出異常

  1. public class AnyOfFutureTest { 
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  3.  
  4.         CompletableFuture<Void> a = CompletableFuture.runAsync(()->{ 
  5.             try { 
  6.                 Thread.sleep(3000L); 
  7.             } catch (InterruptedException e) { 
  8.                 e.printStackTrace(); 
  9.             } 
  10.             System.out.println("我執(zhí)行完了"); 
  11.         }); 
  12.         CompletableFuture<Void> b = CompletableFuture.runAsync(() -> { 
  13.             System.out.println("我也執(zhí)行完了"); 
  14.         }); 
  15.         CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b).whenComplete((m,k)->{ 
  16.             System.out.println("finish"); 
  17. //            return "撿田螺的小男孩"; 
  18.         }); 
  19.         anyOfFuture.join(); 
  20.     } 
  21. } 
  22. //輸出 
  23. 我也執(zhí)行完了 
  24. finish 

thenCompose

thenCompose方法會(huì)在某個(gè)任務(wù)執(zhí)行完成后，將該任務(wù)的執(zhí)行結(jié)果,作為方法入?yún)?去執(zhí)行指定的方法。該方法會(huì)返回一個(gè)新的CompletableFuture實(shí)例

  • 如果該CompletableFuture實(shí)例的result不為null，則返回一個(gè)基于該result新的CompletableFuture實(shí)例;
  • 如果該CompletableFuture實(shí)例為null，然后就執(zhí)行這個(gè)新任務(wù)
  1. public class ThenComposeTest { 
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException { 
  3.  
  4.         CompletableFuture<String> f = CompletableFuture.completedFuture("第一個(gè)任務(wù)"); 
  5.         //第二個(gè)異步任務(wù) 
  6.         ExecutorService executor = Executors.newSingleThreadExecutor(); 
  7.         CompletableFuture<String> future = CompletableFuture 
  8.                 .supplyAsync(() -> "第二個(gè)任務(wù)", executor) 
  9.                 .thenComposeAsync(data -> { 
  10.                     System.out.println(data); return f; //使用第一個(gè)任務(wù)作為返回 
  11.                 }, executor); 
  12.         System.out.println(future.join()); 
  13.         executor.shutdown(); 
  14.  
  15.     } 
  16. } 
  17. //輸出 
  18. 第二個(gè)任務(wù) 
  19. 第一個(gè)任務(wù) 

CompletableFuture使用有哪些注意點(diǎn)

CompletableFuture 使我們的異步編程更加便利的、代碼更加優(yōu)雅的同時(shí)，我們也要關(guān)注下它，使用的一些注意點(diǎn)。

1. Future需要獲取返回值，才能獲取異常信息

  1. ExecutorService executorService = new ThreadPoolExecutor(5, 10, 5L, 
  2.     TimeUnit.SECONDS, new ArrayBlockingQueue<>(10)); 
  3. CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> { 
  4.       int a = 0; 
  5.       int b = 666; 
  6.       int c = b / a; 
  7.       return true; 
  8.    },executorService).thenAccept(System.out::println); 
  9.     
  10.  //如果不加 get()方法這一行，看不到異常信息 
  11.  //future.get(); 

Future需要獲取返回值，才能獲取到異常信息。如果不加 get()/join()方法，看不到異常信息。小伙伴們使用的時(shí)候，注意一下哈,考慮是否加try...catch...或者使用exceptionally方法。

2. CompletableFuture的get()方法是阻塞的。

CompletableFuture的get()方法是阻塞的，如果使用它來(lái)獲取異步調(diào)用的返回值，需要添加超時(shí)時(shí)間~

  1. //反例 
  2.  CompletableFuture.get(); 
  3. //正例 
  4. CompletableFuture.get(5, TimeUnit.SECONDS); 

3. 默認(rèn)線程池的注意點(diǎn)

CompletableFuture代碼中又使用了默認(rèn)的線程池，處理的線程個(gè)數(shù)是電腦CPU核數(shù)-1。在大量請(qǐng)求過(guò)來(lái)的時(shí)候，處理邏輯復(fù)雜的話，響應(yīng)會(huì)很慢。一般建議使用自定義線程池，優(yōu)化線程池配置參數(shù)。

4. 自定義線程池時(shí)，注意飽和策略

CompletableFuture的get()方法是阻塞的，我們一般建議使用future.get(3, TimeUnit.SECONDS)。并且一般建議使用自定義線程池。

但是如果線程池拒絕策略是DiscardPolicy或者DiscardOldestPolicy，當(dāng)線程池飽和時(shí)，會(huì)直接丟棄任務(wù)，不會(huì)拋棄異常。因此建議，CompletableFuture線程池策略最好使用AbortPolicy，然后耗時(shí)的異步線程，做好線程池隔離哈。

參考資料

[1]Java8 CompletableFuture 用法全解: https://blog.csdn.net/qq_31865983/article/details/106137777

[2]基礎(chǔ)篇：異步編程不會(huì)?我教你啊!: https://juejin.cn/post/6902655550031413262#heading-5

[3]CompletableFuture get方法一直阻塞或拋出TimeoutException: https://blog.csdn.net/xiaolyuh123/article/details/85023269

[4]編程老司機(jī)帶你玩轉(zhuǎn) CompletableFuture 異步編程: https://zhuanlan.zhihu.com/p/111841508

[5]解決CompletableFuture異常阻塞: https://blog.csdn.net/weixin_42742643/article/details/111638260

本文轉(zhuǎn)載自微信公眾號(hào)「撿田螺的小男孩」，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系撿田螺的小男孩公眾號(hào)。

 

責(zé)任編輯：武曉燕 來(lái)源： 撿田螺的小男孩
異步編程CompletableFuture
相關(guān)推薦

2024-04-18 08:20:27

Java 8編程工具

2022-07-08 14:14:04

并發(fā)編程異步編程

2021-02-21 14:35:29

Java 8異步編程

2020-05-29 07:20:00

Java8異步編程源碼解讀

2024-03-06 08:13:33

FutureJDKCallable

2016-09-07 20:43:36

Javascript異步編程

2025-02-06 16:51:30

2017-12-21 15:48:11

JavaCompletable

2025-02-28 09:20:00

Future開(kāi)發(fā)代碼

2024-12-26 12:59:39

2023-07-19 08:03:05

Future異步JDK

2024-10-14 08:29:14

異步編程任務(wù)

2024-04-30 11:11:33

aiohttp模塊編程

2015-06-16 11:06:42

JavaCompletable

2023-11-06 08:14:51

Go語(yǔ)言Context

2024-08-06 09:43:54

Java 8工具編程

2013-04-01 15:38:54

異步編程異步編程模型

2023-11-24 16:13:05

C++編程

2025-04-30 01:50:00

C#異步編程

2020-10-15 13:29:57

javascript
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)