日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
實(shí)現(xiàn)異步編程，這個工具類你得掌握！

 [[393773]]

本文轉(zhuǎn)載自微信公眾號「月伴飛魚」，作者日常加油站。轉(zhuǎn)載本文請聯(lián)系月伴飛魚公眾號。

我們提供的服務(wù)有：成都網(wǎng)站設(shè)計、成都網(wǎng)站制作、外貿(mào)網(wǎng)站建設(shè)、微信公眾號開發(fā)、網(wǎng)站優(yōu)化、網(wǎng)站認(rèn)證、韶山ssl等。為上千企事業(yè)單位解決了網(wǎng)站和推廣的問題。提供周到的售前咨詢和貼心的售后服務(wù)，是有科學(xué)管理、有技術(shù)的韶山網(wǎng)站制作公司

前言

最近看公司代碼，多線程編程用的比較多，其中有對CompletableFuture的使用，所以想寫篇文章總結(jié)下

在日常的Java8項(xiàng)目開發(fā)中，CompletableFuture是很強(qiáng)大的并行開發(fā)工具，其語法貼近java8的語法風(fēng)格，與stream一起使用也能大大增加代碼的簡潔性

大家可以多應(yīng)用到工作中，提升接口性能，優(yōu)化代碼

基本介紹

CompletableFuture是Java 8新增的一個類，用于異步編程，繼承了Future和CompletionStage

這個Future主要具備對請求結(jié)果獨(dú)立處理的功能，CompletionStage用于實(shí)現(xiàn)流式處理，實(shí)現(xiàn)異步請求的各個階段組合或鏈?zhǔn)教幚恚虼薱ompletableFuture能實(shí)現(xiàn)整個異步調(diào)用接口的扁平化和流式處理，解決原有Future處理一系列鏈?zhǔn)疆惒秸埱髸r的復(fù)雜編碼

Future的局限性

1、Future 的結(jié)果在非阻塞的情況下，不能執(zhí)行更進(jìn)一步的操作

我們知道，使用Future時只能通過isDone()方法判斷任務(wù)是否完成，或者通過get()方法阻塞線程等待結(jié)果返回，它不能非阻塞的情況下，執(zhí)行更進(jìn)一步的操作。

2、不能組合多個Future的結(jié)果

假設(shè)你有多個Future異步任務(wù)，你希望最快的任務(wù)執(zhí)行完時，或者所有任務(wù)都執(zhí)行完后，進(jìn)行一些其他操作

3、多個Future不能組成鏈?zhǔn)秸{(diào)用

當(dāng)異步任務(wù)之間有依賴關(guān)系時，F(xiàn)uture不能將一個任務(wù)的結(jié)果傳給另一個異步任務(wù)，多個Future無法創(chuàng)建鏈?zhǔn)降墓ぷ髁鳌?/p>

4、沒有異常處理

現(xiàn)在使用CompletableFuture能幫助我們完成上面的事情，讓我們編寫更強(qiáng)大、更優(yōu)雅的異步程序

基本使用

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

通?？梢允褂孟旅鎺讉€CompletableFuture的靜態(tài)方法創(chuàng)建一個異步任務(wù)

 
 
 
 
    
  
  
  
  
  1. public static CompletableFuture runAsync(Runnable runnable);              //創(chuàng)建無返回值的異步任務(wù)
    2. 
  
  
  
  
  2. public static CompletableFuture runAsync(Runnable runnable, Executor executor);     //無返回值，可指定線程池（默認(rèn)使用ForkJoinPool.commonPool）
    3. 
  
  
  
  
  3. public static  CompletableFuture supplyAsync(Supplier supplier);           //創(chuàng)建有返回值的異步任務(wù)
    4. 
  
  
  
  
  4. public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor); //有返回值，可指定線程池
    5.

使用示例：

 
 
 
 
    
  
  
  
  
  1. Executor executor = Executors.newFixedThreadPool(10);
    2. 
  
  
  
  
  2. CompletableFuture future = CompletableFuture.runAsync(() -> {
    3. 
  
  
  
  
  3.     //do something
    4. 
  
  
  
  
  4. }, executor);
    5. 
  
  
  
  
  5. int poiId = 111;
    6. 
  
  
  
  
  6. CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    7. 
  
  
  
  
  7.  PoiDTO poi = poiService.loadById(poiId);
    8. 
  
  
  
  
  8.   return poi.getName();
    9. 
  
  
  
  
  9. });
    10. 
  
  
  
  
  10. // Block and get the result of the Future
    11. 
  
  
  
  
  11. String poiName = future.get();
    12.

使用回調(diào)方法

通過future.get()方法獲取異步任務(wù)的結(jié)果，還是會阻塞的等待任務(wù)完成

CompletableFuture提供了幾個回調(diào)方法，可以不阻塞主線程，在異步任務(wù)完成后自動執(zhí)行回調(diào)方法中的代碼

 
 
 
 
    
  
  
  
  
  1. public CompletableFuture thenRun(Runnable runnable);            //無參數(shù)、無返回值
    2. 
  
  
  
  
  2. public CompletableFuture thenAccept(Consumer action);         //接受參數(shù)，無返回值
    3. 
  
  
  
  
  3. public  CompletableFuture thenApply(Function fn); //接受參數(shù)T，有返回值U
    4.

使用示例：

 
 
 
 
    
  
  
  
  
  1. CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
    2. 
  
  
  
  
  2.                            .thenRun(() -> System.out.println("do other things. 比如異步打印日志或發(fā)送消息"));
    3. 
  
  
  
  
  3. //如果只想在一個CompletableFuture任務(wù)執(zhí)行完后，進(jìn)行一些后續(xù)的處理，不需要返回值，那么可以用thenRun回調(diào)方法來完成。
    4. 
  
  
  
  
  4. //如果主線程不依賴thenRun中的代碼執(zhí)行完成，也不需要使用get()方法阻塞主線程。
    5. 
 
 
 
 
 
 
 
 
    
  
  
  
  
  1. CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
    2. 
  
  
  
  
  2.                            .thenAccept((s) -> System.out.println(s + " world"));
    3. 
  
  
  
  
  3. //輸出：Hello world
    4. 
  
  
  
  
  4. //回調(diào)方法希望使用異步任務(wù)的結(jié)果，并不需要返回值，那么可以使用thenAccept方法
    5. 
 
 
 
 
 
 
 
 
    
  
  
  
  
  1. CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    2. 
  
  
  
  
  2.   PoiDTO poi = poiService.loadById(poiId);
    3. 
  
  
  
  
  3.   return poi.getMainCategory();
    4. 
  
  
  
  
  4. }).thenApply((s) -> isMainPoi(s));   // boolean isMainPoi(int poiId);
    5. 
  
  
  
  
  5. 
  
  
  
  
  6. future.get();
    7. 
  
  
  
  
  7. //希望將異步任務(wù)的結(jié)果做進(jìn)一步處理，并需要返回值，則使用thenApply方法。
    8. 
  
  
  
  
  8. //如果主線程要獲取回調(diào)方法的返回，還是要用get()方法阻塞得到
    9.

組合兩個異步任務(wù)

 
 
 
 
    
  
  
  
  
  1. //thenCompose方法中的異步任務(wù)依賴調(diào)用該方法的異步任務(wù)
    2. 
  
  
  
  
  2. public  CompletableFuture thenCompose(Function> fn); 
    3. 
  
  
  
  
  3. //用于兩個獨(dú)立的異步任務(wù)都完成的時候
    4. 
  
  
  
  
  4. public  CompletableFuture thenCombine(CompletionStage other, 
    5. 
  
  
  
  
  5.                                               BiFunction fn); 
    6.

使用示例：

 
 
 
 
    
  
  
  
  
  1. CompletableFuture> poiFuture = CompletableFuture.supplyAsync(
    2. 
  
  
  
  
  2.   () -> poiService.queryPoiIds(cityId, poiId)
    3. 
  
  
  
  
  3. );
    4. 
  
  
  
  
  4. //第二個任務(wù)是返回CompletableFuture的異步方法
    5. 
  
  
  
  
  5. CompletableFuture> getDeal(List poiIds){
    6. 
  
  
  
  
  6.   return CompletableFuture.supplyAsync(() ->  poiService.queryPoiIds(poiIds));
    7. 
  
  
  
  
  7. }
    8. 
  
  
  
  
  8. //thenCompose
    9. 
  
  
  
  
  9. CompletableFuture> resultFuture = poiFuture.thenCompose(poiIds -> getDeal(poiIds));
    10. 
  
  
  
  
  10. resultFuture.get();
    11.

thenCompose和thenApply的功能類似，兩者區(qū)別在于thenCompose接受一個返回CompletableFuture的Function，當(dāng)想從回調(diào)方法返回的CompletableFuture中直接獲取結(jié)果U時，就用thenCompose

如果使用thenApply，返回結(jié)果resultFuture的類型是CompletableFuture>>，而不是CompletableFuture>

 
 
 
 
    
  
  
  
  
  1. CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
    2. 
  
  
  
  
  2.   .thenCombine(CompletableFuture.supplyAsync(() -> "world"), (s1, s2) -> s1 + s2);
    3. 
  
  
  
  
  3. //future.get()
    4.

組合多個CompletableFuture

當(dāng)需要多個異步任務(wù)都完成時，再進(jìn)行后續(xù)處理，可以使用allOf方法

 
 
 
 
    
  
  
  
  
  1. CompletableFuture poiIDTOFuture = CompletableFuture
    2. 
  
  
  
  
  2.  .supplyAsync(() -> poiService.loadPoi(poiId))
    3. 
  
  
  
  
  3.   .thenAccept(poi -> {
    4. 
  
  
  
  
  4.     model.setModelTitle(poi.getShopName());
    5. 
  
  
  
  
  5.     //do more thing
    6. 
  
  
  
  
  6.   });
    7. 
  
  
  
  
  7. 
  
  
  
  
  8. CompletableFuture productFuture = CompletableFuture
    9. 
  
  
  
  
  9.  .supplyAsync(() -> productService.findAllByPoiIdOrderByUpdateTimeDesc(poiId))
    10. 
  
  
  
  
  10.   .thenAccept(list -> {
    11. 
  
  
  
  
  11.     model.setDefaultCount(list.size());
    12. 
  
  
  
  
  12.     model.setMoreDesc("more");
    13. 
  
  
  
  
  13.   });
    14. 
  
  
  
  
  14. //future3等更多異步任務(wù)，這里就不一一寫出來了
    15. 
  
  
  
  
  15. 
  
  
  
  
  16. CompletableFuture.allOf(poiIDTOFuture, productFuture, future3, ...).join();  //allOf組合所有異步任務(wù)，并使用join獲取結(jié)果
    17.

該方法挺適合C端的業(yè)務(wù)，比如通過poiId異步的從多個服務(wù)拿門店信息，然后組裝成自己需要的模型，最后所有門店信息都填充完后返回

這里使用了join方法獲取結(jié)果，它和get方法一樣阻塞的等待任務(wù)完成

多個異步任務(wù)有任意一個完成時就返回結(jié)果，可以使用anyOf方法

 
 
 
 
    
  
  
  
  
  1. CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
    2. 
  
  
  
  
  2.     try {
    3. 
  
  
  
  
  3.         TimeUnit.SECONDS.sleep(2);
    4. 
  
  
  
  
  4.     } catch (InterruptedException e) {
    5. 
  
  
  
  
  5.        throw new IllegalStateException(e);
    6. 
  
  
  
  
  6.     }
    7. 
  
  
  
  
  7.     return "Result of Future 1";
    8. 
  
  
  
  
  8. });
    9. 
  
  
  
  
  9. 
  
  
  
  
  10. CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
    11. 
  
  
  
  
  11.     try {
    12. 
  
  
  
  
  12.         TimeUnit.SECONDS.sleep(1);
    13. 
  
  
  
  
  13.     } catch (InterruptedException e) {
    14. 
  
  
  
  
  14.        throw new IllegalStateException(e);
    15. 
  
  
  
  
  15.     }
    16. 
  
  
  
  
  16.     return "Result of Future 2";
    17. 
  
  
  
  
  17. });
    18. 
  
  
  
  
  18. 
  
  
  
  
  19. CompletableFuture future3 = CompletableFuture.supplyAsync(() -> {
    20. 
  
  
  
  
  20.     try {
    21. 
  
  
  
  
  21.         TimeUnit.SECONDS.sleep(3);
    22. 
  
  
  
  
  22.     } catch (InterruptedException e) {
    23. 
  
  
  
  
  23.        throw new IllegalStateException(e);
    24. 
  
  
  
  
  24.       return "Result of Future 3";
    25. 
  
  
  
  
  25. });
    26. 
  
  
  
  
  26. 
  
  
  
  
  27. CompletableFuture anyOfFuture = CompletableFuture.anyOf(future1, future2, future3);
  
  
  
  
  28. 
  
  
  
  
  29. System.out.println(anyOfFuture.get()); // Result of Future 2
    30. 
 
 
 
 
    異常處理
    

     
 
 
 
      
  
  
  
  
    1. Integer age = -1;
      2. 
  
  
  
  
    2. 
  
  
  
  
    3. CompletableFuture maturityFuture = CompletableFuture.supplyAsync(() -> {
      4. 
  
  
  
  
    4.   if(age < 0) {
      5. 
  
  
  
  
    5.     throw new IllegalArgumentException("Age can not be negative");
      6. 
  
  
  
  
    6.   }
      7. 
  
  
  
  
    7.   if(age > 18) {
      8. 
  
  
  
  
    8.     return "Adult";
      9. 
  
  
  
  
    9.   } else {
      10. 
  
  
  
  
    10.     return "Child";
      11. 
  
  
  
  
    11.   }
      12. 
  
  
  
  
    12. }).exceptionally(ex -> {
      13. 
  
  
  
  
    13.   System.out.println("Oops! We have an exception - " + ex.getMessage());
      14. 
  
  
  
  
    14.   return "Unknown!";
      15. 
  
  
  
  
    15. }).thenAccept(s -> System.out.print(s));
      16. 
  
  
  
  
    16. //Unkown!
      17. 
 
 
 
 
    exceptionally方法可以處理異步任務(wù)的異常，在出現(xiàn)異常時，給異步任務(wù)鏈一個從錯誤中恢復(fù)的機(jī)會，可以在這里記錄異?；蚍祷匾粋€默認(rèn)值
    使用handler方法也可以處理異常，并且無論是否發(fā)生異常它都會被調(diào)用
     
 
 
 
      
  
  
  
  
    1. Integer age = -1;
      2. 
  
  
  
  
    2. 
  
  
  
  
    3. CompletableFuture maturityFuture = CompletableFuture.supplyAsync(() -> {
      4. 
  
  
  
  
    4.     if(age < 0) {
      5. 
  
  
  
  
    5.         throw new IllegalArgumentException("Age can not be negative");
      6. 
  
  
  
  
    6.     }
      7. 
  
  
  
  
    7.     if(age > 18) {
      8. 
  
  
  
  
    8.         return "Adult";
      9. 
  
  
  
  
    9.     } else {
      10. 
  
  
  
  
    10.         return "Child";
      11. 
  
  
  
  
    11.     }
      12. 
  
  
  
  
    12. }).handle((res, ex) -> {
      13. 
  
  
  
  
    13.     if(ex != null) {
      14. 
  
  
  
  
    14.         System.out.println("Oops! We have an exception - " + ex.getMessage());
      15. 
  
  
  
  
    15.         return "Unknown!";
      16. 
  
  
  
  
    16.     }
      17. 
  
  
  
  
    17.     return res;
      18. 
  
  
  
  
    18. });
      19. 
 
 
 
 
    分片處理
    

    分片和并行處理：分片借助stream實(shí)現(xiàn)，然后通過CompletableFuture實(shí)現(xiàn)并行執(zhí)行，最后做數(shù)據(jù)聚合(其實(shí)也是stream的方法)
    CompletableFuture并不提供單獨(dú)的分片api，但可以借助stream的分片聚合功能實(shí)現(xiàn)
    舉個例子：
     
 
 
 
      
  
  
  
  
    1. //請求商品數(shù)量過多時，做分批異步處理
      2. 
  
  
  
  
    2. List> skuBaseIdsList = ListUtils.partition(skuIdList, 10);//分片
      3. 
  
  
  
  
    3. //并行
      4. 
  
  
  
  
    4. List>> futureList = Lists.newArrayList();
      5. 
  
  
  
  
    5. for (List skuId : skuBaseIdsList) {
      6. 
  
  
  
  
    6.   CompletableFuture> tmpFuture = getSkuSales(skuId);
      7. 
  
  
  
  
    7.   futureList.add(tmpFuture);
      8. 
  
  
  
  
    8. }
      9. 
  
  
  
  
    9. //聚合
      10. 
  
  
  
  
    10. futureList.stream().map(CompletalbleFuture::join).collent(Collectors.toList());
      11. 
 
 
 
 
    舉個例子
    

    帶大家領(lǐng)略下CompletableFuture異步編程的優(yōu)勢
    這里我們用CompletableFuture實(shí)現(xiàn)水泡茶程序
    首先還是需要先完成分工方案，在下面的程序中，我們分了3個任務(wù)：
    

      

    • 任務(wù)1負(fù)責(zé)洗水壺、燒開水
      • 

    • 任務(wù)2負(fù)責(zé)洗茶壺、洗茶杯和拿茶葉
      • 

    • 任務(wù)3負(fù)責(zé)泡茶。其中任務(wù)3要等待任務(wù)1和任務(wù)2都完成后才能開始
      • 

    

    下面是代碼實(shí)現(xiàn)，你先略過runAsync()、supplyAsync()、thenCombine()這些不太熟悉的方法，從大局上看，你會發(fā)現(xiàn)：
    

      

    • 無需手工維護(hù)線程，沒有繁瑣的手工維護(hù)線程的工作，給任務(wù)分配線程的工作也不需要我們關(guān)注;
      • 

    • 語義更清晰，例如 f3 = f1.thenCombine(f2, ()->{}) 能夠清晰地表述任務(wù)3要等待任務(wù)1和任務(wù)2都完成后才能開始;
      • 

    • 代碼更簡練并且專注于業(yè)務(wù)邏輯，幾乎所有代碼都是業(yè)務(wù)邏輯相關(guān)的
      • 

    

     
 
 
 
      
  
  
  
  
    1. //任務(wù)1：洗水壺->燒開水
      2. 
  
  
  
  
    2. CompletableFuture f1 = 
      3. 
  
  
  
  
    3.   CompletableFuture.runAsync(()->{
      4. 
  
  
  
  
    4.   System.out.println("T1:洗水壺...");
      5. 
  
  
  
  
    5.   sleep(1, TimeUnit.SECONDS);
      6. 
  
  
  
  
    6. 
  
  
  
  
    7.   System.out.println("T1:燒開水...");
      8. 
  
  
  
  
    8.   sleep(15, TimeUnit.SECONDS);
      9. 
  
  
  
  
    9. });
      10. 
  
  
  
  
    10. //任務(wù)2：洗茶壺->洗茶杯->拿茶葉
      11. 
  
  
  
  
    11. CompletableFuture f2 = 
      12. 
  
  
  
  
    12.   CompletableFuture.supplyAsync(()->{
      13. 
  
  
  
  
    13.   System.out.println("T2:洗茶壺...");
      14. 
  
  
  
  
    14.   sleep(1, TimeUnit.SECONDS);
      15. 
  
  
  
  
    15. 
  
  
  
  
    16.   System.out.println("T2:洗茶杯...");
      17. 
  
  
  
  
    17.   sleep(2, TimeUnit.SECONDS);
      18. 
  
  
  
  
    18. 
  
  
  
  
    19.   System.out.println("T2:拿茶葉...");
      20. 
  
  
  
  
    20.   sleep(1, TimeUnit.SECONDS);
      21. 
  
  
  
  
    21.   return "龍井";
      22. 
  
  
  
  
    22. });
      23. 
  
  
  
  
    23. //任務(wù)3：任務(wù)1和任務(wù)2完成后執(zhí)行：泡茶
      24. 
  
  
  
  
    24. CompletableFuture f3 = 
      25. 
  
  
  
  
    25.   f1.thenCombine(f2, (__, tf)->{
      26. 
  
  
  
  
    26.     System.out.println("T1:拿到茶葉:" + tf);
      27. 
  
  
  
  
    27.     System.out.println("T1:泡茶...");
      28. 
  
  
  
  
    28.     return "上茶:" + tf;
      29. 
  
  
  
  
    29.   });
      30. 
  
  
  
  
    30. //等待任務(wù)3執(zhí)行結(jié)果
      31. 
  
  
  
  
    31. System.out.println(f3.join());
      32. 
  
  
  
  
    32. 
  
  
  
  
    33. void sleep(int t, TimeUnit u) {
      34. 
  
  
  
  
    34.   try {
      35. 
  
  
  
  
    35.     u.sleep(t);
      36. 
  
  
  
  
    36.   }catch(InterruptedException e){}
      37. 
  
  
  
  
    37. }
      38. 
 
 
 
 
    注意事項(xiàng)
    

    1.CompletableFuture默認(rèn)線程池是否滿足使用
    前面提到創(chuàng)建CompletableFuture異步任務(wù)的靜態(tài)方法runAsync和supplyAsync等，可以指定使用的線程池，不指定則用CompletableFuture的默認(rèn)線程池
     
 
 
 
      
  
  
  
  
    1. private static final Executor asyncPool = useCommonPool ?
      2. 
  
  
  
  
    2.         ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
      3. 
 
 
 
 
    可以看到，CompletableFuture默認(rèn)線程池是調(diào)用ForkJoinPool的commonPool()方法創(chuàng)建，這個默認(rèn)線程池的核心線程數(shù)量根據(jù)CPU核數(shù)而定，公式為Runtime.getRuntime().availableProcessors() - 1，以4核雙槽CPU為例，核心線程數(shù)量就是4*2-1=7個
    這樣的設(shè)置滿足CPU密集型的應(yīng)用，但對于業(yè)務(wù)都是IO密集型的應(yīng)用來說，是有風(fēng)險的，當(dāng)qps較高時，線程數(shù)量可能就設(shè)的太少了，會導(dǎo)致線上故障
    所以可以根據(jù)業(yè)務(wù)情況自定義線程池使用
    2.get設(shè)置超時時間不能串行g(shù)et，不然會導(dǎo)致接口延時線程數(shù)量*超時時間
                                                    
    
                                                新聞標(biāo)題：實(shí)現(xiàn)異步編程，這個工具類你得掌握！                                                
    
                                                當(dāng)前網(wǎng)址：http://www.dlmjj.cn/article/djojgoo.html