Java8系列 (七) CompletableFuture異步編程

本文轉載自查看原文 2019-11-08 08:48 1088 Java8

概述

Java8之前用 Future 處理異步請求, 當你需要獲取任務結果時, 通常的做法是調用 get(long timeout, TimeUnit unit) 此方法會阻塞當前的線程, 如果任務處理超時, 就會拋出一個 TimeoutException

    @Test
    public void test1() throws InterruptedException, ExecutionException, TimeoutException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Future<String> f = executorService.submit(() -> "ceshishanghu");
        String s = f.get(3, TimeUnit.SECONDS);
        System.out.println(s);
    }

在Java8中引入了 CompletableFuture, 使用它提供的API可以不用像之前那樣阻塞式或輪詢的獲取某個異步任務的結果, CompletableFuture 會在異步任務處理完成后自動進行回調, 讓你可以鏈式的組合多個異步任務。

CompletableFuture 類中提供了許多以 Async 后綴結尾的方法。通常而言，名稱中不帶 Async 的方法和它的前一個任務一樣，在同一個線程中運行。而名稱以 Async 結尾的方法會將后續的任務提交到一個線程池，所以每個任務是由不同的線程處理的。

靜態工廠方法

supplyAsync(): 異步處理任務, 有返回值
runAsync(): 異步處理任務, 沒有返回值
allOf(): 需要等待所有的異步任務都執行完畢,才會返回一個新的CompletableFuture
anyOf(): 任意一個異步任務執行完畢,就會返回一個新的CompletableFuture
completedFuture(): 這種方式獲取的 CompletableFuture 不是異步的，它會等待獲取明確的返回結果之后再返回一個已經完成的 CompletableFuture

    @Test
    public void test2() {
        //創建一個已經有任務結果的CompletableFuture
        CompletableFuture<String> f1 = CompletableFuture.completedFuture("return value");
        //異步處理任務,有返回值
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);
        //異步處理任務,沒有返回值
        CompletableFuture<Void> f3 = CompletableFuture.runAsync(System.out::println);
        //需要等待所有的異步任務都執行完畢,才會返回一個新的CompletableFuture
//        CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);
        //任意一個異步任務執行完畢,就會返回一個新的CompletableFuture
        CompletableFuture<Object> any = CompletableFuture.anyOf(f1, f2, f3);
        Object result = any.join();
        System.out.println("result = " + result);//result = return value
    }

    public String get() {
        delay();
        return "異步任務結果";
    }

    public void delay() {
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

上面的示例中, allOf() 因為要等待所有的異步任務執行完成，所以要延時1秒鍾才會返回一個新的 CompletableFuture, 而 anyOf() 則不需要等待所有的異步任務, 因為第一個異步最先完成, 所以控制台輸出 result = return value 。

鏈式調用

A任務執行完畢, 繼續執行B任務, B任務執行完畢, 繼續執行C任務...

    @Test
    public void test2() {
        CompletableFuture<Void> f = CompletableFuture.supplyAsync(() -> {
            //測試拋異常后,handle()方法接受並處理
　　　　　　　//int x = 1 / 0;
            return "這是一個栗子";
        }).handle((res, ex) -> {
            System.out.println("handle res = " + res);
            if (Objects.nonNull(ex)) {
                System.out.println("handle ex" + ex.getCause().getMessage());
            }
            return Objects.nonNull(ex) ? 0 : 1;
        }).thenApply(res -> {
            System.out.println("thenApply res = " + res);
            return res == 1 ? "success" : "error";
        }).thenAccept(res -> System.out.println("thenAccept res = " + res)
        ).thenRun(() -> System.out.println("沒有參數, 異步執行一個沒有返回值的任務"));
        f.join();
    }

輸出結果:

handle res = 這是一個栗子
thenApply res = 1
thenAccept res = success
沒有參數, 異步執行一個沒有返回值的任務

將上面 int x = 1 / 0; 這行代碼取消注釋, 重新運行結果如下:

handle res = null
handle ex/ by zero
thenApply res = 0
thenAccept res = error
沒有參數, 異步執行一個沒有返回值的任務

可以看到, handle() 方法接受前一個 CompletableFuture 的返回結果或拋出的異常作為方法入參, 經過處理后再返回一個新的結果。

級聯組合

thenCompose(): 對兩個異步操作進行組合，第一個操作完成時，將其結果作為參數傳遞給第二個操作, 第二個操作會返回一個新的CompletableFuture。
thenCombine(): 將兩個完全無關聯的異步請求的結果整合起來, 計算出一個新的值並返回

    @Test
    public void test3() {
        CompletableFuture<String> f = CompletableFuture.completedFuture("CompletableFuture 1");
        CompletableFuture<String> f1 = f.thenCompose(res -> {
            System.out.println("thenCompose res = " + res);
            return CompletableFuture.supplyAsync(() -> "CompletableFuture 2");
        });
        System.out.println(f1.join());
        CompletableFuture<Integer> f3 = CompletableFuture.completedFuture(998);
        CompletableFuture<String> f4 = f.thenCombine(f3, (str, num) -> {
            System.out.println("str = " + str + ", num= " + num);
            return str + num;
        });
        System.out.println(f4.join());
    }

輸出結果:

thenCompose res = CompletableFuture 1
CompletableFuture 2
str = CompletableFuture 1, num= 998
CompletableFuture 1998

whenComplete

當前一個 CompletableFuture 計算完成或拋出異常時, 可以使用 whenComplete() 執行指定的任務。

    @Test
    public void test4() {
        CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
            //測試拋異常后,whenComplete()方法接受並處理
            int x = 1 / 0;
            return "這是一個栗子";
        }).whenComplete((res, ex) -> {
            System.out.println("whenComplete res = " + res);
            if (Objects.nonNull(ex)) {
                System.out.println("whenComplete ex" + ex.getCause().getMessage());
            }
        });
        System.out.println("f.join() = " + f.join());
    }

輸出結果如下，其中 res 對應前一個 CompletableFuture 的返回結果，ex 對應前一個 CompletableFuture 拋出的異常(如果發生異常)。

從控制台輸出順序看出，當前一個 CompletableFuture 計算完成或拋出異常時, whenComplete() 會接受它的返回結果或拋出的異常，來做一些其他的事情，最后再返回原來的返回結果或拋出異常。類比下 try/catch 語句塊中的 final 語句塊。

whenComplete res = null
whenComplete ex/ by zero

java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero

    at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
    at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
    at java.util.concurrent.CompletableFuture$AsyncSupply.run$$$capture(CompletableFuture.java:1592)
    at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java)
    at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.ArithmeticException: / by zero
    at com.java8.action.ChapterTest.lambda$test4$0(ChapterTest.java:22)

異常處理

只有當前一個 CompletableFuture 發生異常時，才會進入到 exceptionally() 方法，並將產生的異常作為入參。

    @Test
    public void test5() {
        CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
            //測試拋異常后,exceptionally()方法接受並處理
            //int x = 1 / 0;
            return "這是一個栗子";
        }).exceptionally(ex -> ex.getCause().getMessage());
        System.out.println("f.join() = " + f.join());
    }

注釋 int x = 1 / 0; ，輸出如下：

f.join() = 這是一個栗子

取消注釋 int x = 1 / 0; , 輸出如下：

f.join() = / by zero

Both系列方法

thenAcceptBoth(): 等待當前的 CompletableFuture 和另一個 CompletableFuture 執行完成，將它們的返回結果作為入參去執行一個操作，沒有返回值
runAfterBoth(): 等待當前的 CompletableFuture 和另一個 CompletableFuture 執行完成，然后去執行一個操作，沒有返回值

代碼清單一

    @Test
    public void test6() {
        CompletableFuture<Integer> f1 = CompletableFuture.completedFuture(9523);
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);
        CompletableFuture<Void> both = f1.thenAcceptBoth(f2, (num, str) -> System.out.println("num = " + num + ", str = " + str));
        both.join();
    }

    public String get() {
        delay();
        return "CompletableFuture 2";
    }

    public void delay() {
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

代碼清單一輸出結果如下：

num = 9523, str = CompletableFuture 2

代碼清單二

    @Test
    public void test7() {
        CompletableFuture<Integer> f1 = CompletableFuture.completedFuture(9523);
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "");
        CompletableFuture<Void> both = f1.runAfterBoth(f2, () -> System.out.println("執行一個任務，沒有入參"));
        both.join();
    }

代碼清單二輸出結果如下：

執行一個任務，沒有入參

Either系列

acceptEither: 當前的 CompletableFuture 和另一個 CompletableFuture 任意一個執行完成，將對應的返回結果作為入參去執行一個操作，沒有返回值
applyToEither: 當前的 CompletableFuture 和另一個 CompletableFuture 任意一個執行完成，將對應的返回結果作為入參，使用 mapping 函數轉換成一個新的值並返回
runAfterEither: 當前的 CompletableFuture 和另一個 CompletableFuture 任意一個執行完成，然后去執行一個操作，沒有返回值

代碼清單三：

    @Test
    public void test8() {
        CompletableFuture<String> f1 = CompletableFuture.completedFuture("CompletableFuture 1");
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);
        CompletableFuture<Void> both = f1.acceptEither(f2, System.out::println);
        both.join();
    }

    public String get() {
        delay();
        return "CompletableFuture 2";
    }

    public void delay() {
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

代碼清單三輸出結果：

CompletableFuture 1

代碼清單四：

    @Test
    public void test9() {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(this::get);
        CompletableFuture<String> f2 = CompletableFuture.completedFuture("CompletableFuture 2");
        CompletableFuture<Integer> f3 = f1.applyToEither(f2, res -> {
            System.out.println("res = " + res);
            return res.length();
        });
        System.out.println("f3.join() = " + f3.join());
    }

    public String get() {
        delay();//這里會延時一秒鍾
        return "CompletableFuture 1";
    }

代碼清單四輸出結果：

res = CompletableFuture 2
f3.join() = 19

代碼清單五：

    @Test
    public void test10() {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(this::get);
        CompletableFuture<Void> f2 = CompletableFuture.allOf();
        CompletableFuture<Void> f3 = f1.runAfterEither(f2, () -> System.out.println("執行一個任務，沒有入參"));
        f3.join();
    }

    public String get() {
        delay();//這里會延時一秒鍾
        return "CompletableFuture 1";
    }

代碼清單五輸出結果：

執行一個任務，沒有入參

使用自定義的執行器來處理多個異步任務

在實際應用場景中可能會遇到這種情況，假如你需要同時處理大量的異步任務，且這些異步任務互相不依賴，你只要最后把它們的結果組裝起來就行，這該怎么實現呢？

下面給出了一個使用默認執行器的示例，通過Stream流同時創建 9 個異步任務，獲取它們的結果並組裝后返回，其中 Runtime.getRuntime().availableProcessors() 表示Java虛擬機可用的處理器個數，在我之前的文章 Java8系列 (二) Stream流中有介紹過。

代碼清單六：

    @Test
    public void test11() {
        List<String> list = Arrays.asList("王小波書店", "杭州沈記古舊書店", "貓的天空之城概念書店", "純真年代書吧", "南山書屋", "西西弗書店", "新華書店", "鍾書閣", "雲門書屋");
        System.out.println("當前機器有" + Runtime.getRuntime().availableProcessors() + "個可用的處理器");
        long start = System.nanoTime();
        List<CompletableFuture<String>> futures = list.stream()
                .map(str -> CompletableFuture.supplyAsync(() -> this.calculateLength(str)))
                .collect(Collectors.toList());
        System.out.println("get futures "+(System.nanoTime() - start) / 1000_000 + " msecs");
        String result = futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.joining(",", "[", "]"));
        System.out.println("get result "+(System.nanoTime() - start) / 1000_000 + " msecs");
        System.out.println(result);
    }

    public String calculateLength(String str) {
        delay();
        return str;
    }

    public void delay() {
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

運行代碼清單六，輸出結果：

當前機器有4個可用的處理器
get futures 95 msecs
get result 3098 msecs
[王小波書店,杭州沈記古舊書店,貓的天空之城概念書店,純真年代書吧,南山書屋,西西弗書店,新華書店,鍾書閣,雲門書屋]

可以看到，雖然使用了異步處理，但還是花了 3098 毫秒才執行完成所有任務。這是因為 CompletableFuture 內部采用的是通用線程池 ForkJoinPool.commonPool() , 默認都使用固定數目的線程, 具體線程數取決於 Runtime.getRuntime().availableProcessors() 的返回值。

我這里測試的機器顯示通用線程池中處於可用狀態的線程數為 4，一次只能同時處理 4 個任務，后面的5個異步任務只能等到前面某一個操作完成釋放出空閑線程才能繼續, 因此總的會消耗約 3 秒鍾的時間。

我們將上面的代碼進行重構，使用自定義的執行器，通過自定義的執行器你可以指定線程池的大小。其中線程數的設定可以參考公式 Nthreads = NCPU * UCPU * (1 + W/C)

    @Test
    public void test12() {
        List<String> list = Arrays.asList("王小波書店", "杭州沈記古舊書店", "貓的天空之城概念書店", "純真年代書吧", "南山書屋", "西西弗書店", "新華書店", "鍾書閣", "雲門書屋");
        final ExecutorService executor = Executors.newFixedThreadPool(Math.min(list.size(), 100), r -> {
            Thread thread = new Thread(r);
            //守護線程不會組織程序的終止
            thread.setDaemon(true);
            return thread;
        });
        System.out.println("當前機器有" + Runtime.getRuntime().availableProcessors() + "個可用的處理器, 當前處理異步請求的線程池大小為 " + Math.min(list.size(), 100));
        long start = System.nanoTime();
        List<CompletableFuture<String>> futures = list.stream()
                .map(str -> CompletableFuture.supplyAsync(() -> this.calculateLength(str), executor))
                .collect(Collectors.toList());
        System.out.println("get futures " + (System.nanoTime() - start) / 1000_000 + " msecs");
        String result = futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.joining(",", "[", "]"));
        System.out.println("get result " + (System.nanoTime() - start) / 1000_000 + " msecs");
        System.out.println(result);
    }

    public String calculateLength(String str) {
        delay();
        return str;
    }

    public void delay() {
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

輸出結果如下：

當前機器有4個可用的處理器, 當前處理異步請求的線程池大小為 9
get futures 38 msecs
get result 1039 msecs
[王小波書店,杭州沈記古舊書店,貓的天空之城概念書店,純真年代書吧,南山書屋,西西弗書店,新華書店,鍾書閣,雲門書屋]

可以看到，使用自定義的執行器調大線程池大小后，總的運行時間只要 1039 毫秒。

將CompletableFuture作為Controller的返回值

上面還存在一個問題，雖然現在可以同時處理多個異步任務，但是如果需要將異步結果返回給另一個服務，那不是還得通過 join() 阻塞的獲取到返回值后才能再返回么？

自Spring Boot 1.3 (Spring 4.2) 之后開始支持 CompletableFuture 或 CompletionStage 作為 Controller 的返回值，她很好的解決了上面的異步阻塞問題，只要將 CompletableFuture 作為 Controller 的返回值，在異步任務執行完成后，它會自動響應結果給另一個服務。

@RestController
public class AsyncController {

    @GetMapping("/redirect")
    public CompletableFuture<ModelAndView> redirect() {
        return CompletableFuture.supplyAsync(() -> {
            this.delay();
            RedirectView redirectView = new RedirectView("https://www.cnblogs.com/qingshanli/");
            redirectView.addStaticAttribute("hint", "CompletableFuture組裝ModelAndView視圖,異步返回結果");
            return new ModelAndView(redirectView);
        });
    }

    @GetMapping("/async")
    public CompletableFuture<String> async() {
        System.out.println("async method start");
        return CompletableFuture.supplyAsync(() -> {
            this.delay();
            return "CompletableFuture作為Controller的返回值,異步返回結果";
        }).whenComplete((res, ex) -> System.out.println("async method completely, res = " + res + ", ex = " + ex));
    }

    public void delay() {
        try {
            Thread.sleep(3000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

啟動項目，Postman 訪問 http://localhost:8080/async，截圖如下：

Postman 訪問 http://localhost:8080/redirect，截圖如下：

參考資料

https://github.com/AndreasKl/spring-boot-mvc-completablefuture

https://nickebbitt.github.io/blog/2017/03/22/async-web-service-using-completable-future

https://www.humansreadcode.com/spring-boot-completablefuture/

Java8 實戰

作者：張小凡
出處：https://www.cnblogs.com/qingshanli/
本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接，否則保留追究法律責任的權利。如果覺得還有幫助的話，可以點一下右下角的【推薦】。

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 了不起的Java-CompletableFuture組合異步編程【JDK8】Java8 優雅的異步調用API CompletableFuture Java異步CompletableFuture的使用 Java8新特性--CompletableFuture 使用CompletableFuture實現異步編程 Java並發編程系列一：Future和CompletableFuture解析與使用編程老司機帶你玩轉 CompletableFuture 異步編程 java8 CompletableFuture，allOf多實例返回 Java8函數之旅 (八) - 組合式異步編程 Java並發包異步執行器CompletableFuture