Java中的CompletableFuture詳解

更新時間：2023年09月27日 08:32:27 作者：苦?糖?果

這篇文章主要介紹了Java中的CompletableFuture詳解,Future接口(FutueTask實現(xiàn)類)定義了操作異步任務(wù)執(zhí)行一些方法,如獲取異步任務(wù)的執(zhí)行結(jié)果、取消任務(wù)的執(zhí)行、判斷任務(wù)是否被取消、判斷任務(wù)執(zhí)行是否完畢等,需要的朋友可以參考下

1.Future

1.1 Future接口理論知識復(fù)習(xí)

Future接口(FutueTask實現(xiàn)類)定義了操作異步任務(wù)執(zhí)行一些方法，如獲取異步任務(wù)的執(zhí)行結(jié)果、取消任務(wù)的執(zhí)行、判斷任務(wù)是否被取消、判斷任務(wù)執(zhí)行是否完畢等。比如主線程讓一個子線程去執(zhí)行任務(wù)，子線程可能比較耗時，啟動子線程開始執(zhí)行任務(wù)后，主線程就去做其他事情了，忙其它事情或者先執(zhí)行完，過了一會才去獲取子任務(wù)的執(zhí)行結(jié)果或變更的任務(wù)狀態(tài)。

Future接口可以為主線程開一個分支任務(wù)，專門為主線程處理耗時和費(fèi)力的復(fù)雜業(yè)務(wù)。

Future接口能干什么?

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

代碼說話: Runnable接口Callable接口

Future接口和FutureTask實現(xiàn)類

目的:異步多線程任務(wù)執(zhí)行且返回有結(jié)果，三個特點:多線程/有返回/異步任務(wù)

1.2 FutureTask架構(gòu)

綠色虛線：表示實現(xiàn)的關(guān)系，實現(xiàn)一個接口綠色實線：表示接口之間的繼承藍(lán)色實線：表示類之間的繼承

1.3 Future編碼實戰(zhàn)

public class FutureTaskTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        long start = System.currentTimeMillis();
        FutureTask<String> task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(2);
            return "2";
        });
        FutureTask<String> task1 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(1);
            return "1";
        });
        FutureTask<String> task3 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return "3";
        });
        executorService.submit(task1);
        executorService.submit(task2);
        executorService.submit(task3);
        System.out.println(task1.get());
        System.out.println(task2.get(3,TimeUnit.SECONDS));
        while (true){
            if(task3.isDone()){
                System.out.println(task3.get());
                break;
            }else {
                TimeUnit.MILLISECONDS.sleep(200);
            }
        }
        System.out.println("執(zhí)行耗時："+(System.currentTimeMillis()-start));
        executorService.shutdown();
    }
}

1
2
3
執(zhí)行耗時：3066

優(yōu)缺點分析

優(yōu)點

future+線程池異步多線程任務(wù)配合，能顯著提高程序的執(zhí)行效率。

缺點

一旦調(diào)用get()方法求結(jié)果，如果計算沒有完成容易導(dǎo)致程序阻塞 isDone()輪詢的方式會耗費(fèi)無謂的CPU資源,而且也不見得能及時地得到計算結(jié)果. 如果想要異步獲取結(jié)果,通常都會以輪詢的方式去獲取結(jié)果盡量不要阻塞
Future對于結(jié)果的獲取不是很友好，只能通過阻塞或輪詢的方式得到任務(wù)的結(jié)果。
將多個異步任務(wù)的計算結(jié)果組合起來，后一個異步任務(wù)的計算結(jié)果需要前一個異步任務(wù)的值
將兩個或多個異步計算合成一個異步計算，這幾個異步計算互相獨立，同時后面這個又依賴前一個處理的結(jié)果。
對計算速度選最快：當(dāng)Future集合中某個任務(wù)最快結(jié)束時，返回結(jié)果，返回第一名處理

對于簡單的業(yè)務(wù)場景使用Future完全OK,但想完成上述一些復(fù)雜的任務(wù)，使用Future之前提供的那點API就囊中羞澀，處理起來不夠優(yōu)雅，這時候還是讓CompletableFuture以聲明式的方式優(yōu)雅的處理這些需求。Future能干的，CompletableFuture都能干。

2. CompletableFuture

2.1 CompletableFuture對Future的改進(jìn)

CompletableFuture異步線程發(fā)生異常，不會影響主線程，用來記錄日志特別方便。

CompletableFuture為什么出現(xiàn) get()方法在Future 計算完成之前會一直處在阻塞狀態(tài)下，isDone()方法容易耗費(fèi)CPU資源，對于真正的異步處理我們希望是可以通過傳入回調(diào)函數(shù)，在Future結(jié)束時自動調(diào)用該回調(diào)函數(shù)，這樣，我們就不用等待結(jié)果。

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

CompletableFuture和CompletionStage

CompletionStage CompletionStage代表異步計算過程中的某一個階段，一個階段完成以后可能會觸發(fā)另外一個階段一個階段的計算執(zhí)行可以是一個Function, Consumer或者Runnable。比如: stage.thenApply(x -> square(x)).thenAccept(×->System.out.print(x)).thenRun(( ->systeh.out.println()) 一個階段的執(zhí)行可能是被單個階段的完成觸發(fā)，也可能是由多個階段一起觸發(fā) 代表異步計算過程中的某一個階段，一個階段完成以后可能會觸發(fā)另外一個階段，有些類似Linux系統(tǒng)的管道分隔符傳參數(shù)。

CompletableFuture 在Java8中，CompletableFuture提供了非常強(qiáng)大的Future的擴(kuò)展功能，可以幫助我們簡化異步編程的復(fù)雜性，并且提供了函數(shù)式編程的能力，可以通過回調(diào)的方式處理計算結(jié)果，也提供了轉(zhuǎn)換和組合CompletableFuture 的方法。它可能代表一個明確完成的Future，也有可能代表一個完成階段(CompletionStage )，它支持在計算完成以后觸發(fā)一些函數(shù)或執(zhí)行某些動作。它實現(xiàn)了Future和CompletionStage接口

核心的四個靜態(tài)方法，來創(chuàng)建一個異步任務(wù)

從Java8開始引入了CompletableFuture，它是Future的功能增強(qiáng)版。減少阻塞和輪詢，可以傳入回調(diào)對象，當(dāng)異步任務(wù)完成或者發(fā)生異常時，自動調(diào)用回調(diào)對象的回調(diào)方法

CompletableFuture的優(yōu)點異步任務(wù)結(jié)束時，會自動回調(diào)某個對象的方法; 主線程設(shè)置好回調(diào)后，不再關(guān)心異步任務(wù)的執(zhí)行，異步任務(wù)之間可以順序執(zhí)行異步任務(wù)出錯時，會自動回調(diào)某個對象的方法;

2.2 案例精講

函數(shù)式編程已經(jīng)主流先說說join和get對比說說你過去工作中的項目亮點?大廠業(yè)務(wù)需求說明一波流Java8函數(shù)式編程帶走-比價案例實戰(zhàn)

Lambda表達(dá)式+Stream流式調(diào)用+Chain鏈?zhǔn)秸{(diào)用+Java8函數(shù)式編程

案例精講-從電商網(wǎng)站的比價需求講起

需求說明同一款產(chǎn)品，同時搜索出同款產(chǎn)品在各大電商平臺的售價;

輸出返回: 出來結(jié)果希望是同款產(chǎn)品的在不同地方的價格清單列表，返回一個List《mysql》in jd price is 88.05 《mysql》in dangdang price is 86.11 《mysql》in taobao price is 90.43

解決方案，比對同一個商品在各個平臺上的價格，要求獲得一個清單列表，

1 ）step by step，按部就班，查完京東查淘寶，查完淘寶查天貓

2 ）all in，萬箭齊發(fā)，一口氣多線程異步任務(wù)同時查詢

public class CompletableFutureDemo {
    static List<NetMall> list = Arrays.asList(
            new NetMall("vip"),
            new NetMall("jd"),
            new NetMall("tb"),
            new NetMall("pdd")
    );
    public static void main(String[] args) {
        long cur1 = System.currentTimeMillis();
        getPrice("Phone").forEach(r-> System.out.println(r));
        System.out.println("getPrice耗時"+(System.currentTimeMillis()-cur1));
        long cur2 = System.currentTimeMillis();
        getPriceByCompletableFuture("Phone").forEach(r-> System.out.println(r));
        System.out.println("getPriceByCompletableFuture耗時"+(System.currentTimeMillis()-cur2));
    }
    private static List<String> getPrice(String productName){
        return list.stream()
                .map(r->String.format(productName+" in %s price is %.2f",r.getName(),r.calcPrice(productName)))
                .collect(Collectors.toList());
    }
    private static List<String> getPriceByCompletableFuture(String productName){
        return list.stream()
                .map(r-> CompletableFuture.supplyAsync(()->String.format(productName+" in %s price is %.2f",r.getName(),r.calcPrice(productName))))
                .collect(Collectors.toList())
                .stream()
                .map(s->s.join())
                .collect(Collectors.toList());
    }
}
class NetMall{
    private String name;
    public double calcPrice(String productName){
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return  ThreadLocalRandom.current().nextDouble(100000000)+productName.hashCode();
    }
    public NetMall(String name) {
        this.name = name;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
}

2.3 CompletableFuture常用方法

2.3.1 獲得結(jié)果和觸發(fā)計算

獲得結(jié)果

public T get() 不見不散
public T get(long timeout,TimeUnit unit) 過時不候
public T join()
public T getNow(T valuelfAbsent)

沒有計算完成的情況下，給我一個替代結(jié)果。計算完，返回計算完成后的結(jié)果。立即獲取結(jié)果不阻賽。沒算完，返回設(shè)定的valuelfAbsent值

主動觸發(fā)計算 public bgolean complete(T value) 是否打斷get方法立即返回括號值

public class CompletableFutureTest {
  public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
      CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
          try {
              Thread.sleep(2000);
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
          return "hello CompletableFuture";
      });
      System.out.println(completableFuture.getNow("心急吃不了熱豆腐"));
      System.out.println(completableFuture.get());
      System.out.println(completableFuture.get(1500, TimeUnit.MILLISECONDS));
      System.out.println(completableFuture.join());
      System.out.println(completableFuture.complete("未雨綢繆")+"\t"+completableFuture.join());
  }
}

2.3.2 對計算結(jié)果進(jìn)行處理

thenApply 計算結(jié)果存在依賴關(guān)系,這兩個線程串行化異常相關(guān):由于存在依賴關(guān)系(當(dāng)前步錯，不走下一步)，當(dāng)前步驟有異常的話就叫停。

public class CompletableFutureTest2 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 6;
        },executorService).thenApply((r)-> {
            int i=2/0;
            return r * 5;
        }).thenApply((r)-> {
            System.out.println(r);
            return r - 2;
        }).whenComplete((v, e) -> {
            System.out.println("計算結(jié)果："+v);
        }).exceptionally(e -> {
            System.out.println(e.getMessage());
            System.out.println(e);
            return null;
        });
        System.out.println("============主線程==========");
        executorService.shutdown();
    }
}

發(fā)生異常后進(jìn)入exceptionally代碼塊，但是thenApply中的代碼不會執(zhí)行，whenComplete依舊會執(zhí)行

============主線程==========
計算結(jié)果：null
java.lang.ArithmeticException: / by zero
java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero

handle

計算結(jié)果存在依賴關(guān)系，這兩個線程串行化異常相關(guān):有異常也可以往下一步走，根據(jù)帶的異常參數(shù)可以進(jìn)步處理

發(fā)生異常后進(jìn)入exceptionally代碼塊，但是handle和whenComplete依舊會執(zhí)行

============主線程==========
null
計算結(jié)果：null
java.lang.NullPointerException
java.util.concurrent.CompletionException: java.lang.NullPointerException

2.3.3 對計算結(jié)果進(jìn)行消費(fèi)

接收任務(wù)的處理結(jié)果，并消費(fèi)處理，無返回結(jié)果thenAccept

public class CompletableFutureTest3 {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(()->{
            return 3;
        }).thenApply(r->{
            return r*8;
        }).thenApply(r->{
            return r/2;
        }).thenAccept(r-> System.out.println(r));
        System.out.println(CompletableFuture.supplyAsync(()->"6666").thenRun(()->{}).join());
        System.out.println(CompletableFuture.supplyAsync(()->"6666").thenAccept(r-> System.out.println(r)).join());
        System.out.println(CompletableFuture.supplyAsync(()->"6666").thenApply(r->r+"9999").join());
    }
}

12
null
6666
null
66669999

completableFuture和線程池說明

以thenRun和thenRunAsync為例，有什么區(qū)別?

沒有傳入自定義線程池，都用默認(rèn)線程池ForkJoinPool; 傳入了一個自定義線程池，如果你執(zhí)行第一個任務(wù)的時候，傳入了一個自定義線程池: 調(diào)用thenRun方法執(zhí)行第二個任務(wù)時，則第二個任務(wù)和第一個任務(wù)是共用同一個線程池。調(diào)用thenRunAsync執(zhí)行第二個任務(wù)時，則第一個任務(wù)使用的是你自己傳入的線程池，第二個任務(wù)使用的是ForkJoin線程池

有可能處理太快，系統(tǒng)優(yōu)化切換原則，直接使用main線程處理其它如: thenAccept和thenAcceptAsync，thenApply和thenApplyAsync等，它們之間的區(qū)別也是同理

2.3.4 對計算速度進(jìn)行選用與對計算結(jié)果進(jìn)行合并

applyToEither：誰快用誰 thenCombine：兩個completionStage任務(wù)都完成后，最終能把兩個任務(wù)的結(jié)果一起交給thenCombine來處理。先完成的先等著，等待其它分支任務(wù)

public class CompletableFutureTest4 {
    public static void main(String[] args) {
        CompletableFuture<String> first = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "1號選手";
        });
        CompletableFuture<String> second = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "2號選手";
        });
        CompletableFuture<String> result = first.applyToEither(second, r -> r + "is winner");
        CompletableFuture<String> res = first.thenCombine(second, (x, y) -> x + y);
        System.out.println(result.join());
        System.out.println(res.join());
    }
}