CompletableFuture：讓你的代碼免受阻塞之苦

提升應用性能的時候很容易就會想起異步，異步去處理一些任務這樣主線程能夠儘快響應。

寫在前面

經過閱讀本篇文章你將瞭解到：

• CompletableFuture的使用
• CompletableFure異步和同步的性能測試
• 已經有了Future爲何仍須要在JDK1.8中引入CompletableFuture
• 對CompletableFuture的使用優化

場景說明

查詢全部商店某個商品的價格並返回，而且查詢商店某個商品的價格的API爲同步
一個Shop類，提供一個名爲getPrice的同步方法

• 店鋪類：Shop.java

public class Shop {
    private Random random = new Random();
    /**
     * 根據產品名查找價格
     * */
    public double getPrice(String product) {
        return calculatePrice(product);
    }

    /**
     * 計算價格
     *
     * @param product
     * @return
     * */
    private double calculatePrice(String product) {
        delay();
        //random.nextDouble()隨機返回折扣
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }

    /**
     * 經過睡眠模擬其餘耗時操做
     * */
    private void delay() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

查詢商品的價格爲同步方法，並經過sleep方法模擬其餘操做。這個場景模擬了當須要調用第三方API，但第三方提供的是同步API，在沒法修改第三方API時如何設計代碼調用提升應用的性能和吞吐量，這時候可使用CompletableFuture類

CompletableFuture使用

Completable是Future接口的實現類，在JDK1.8中引入

• CompletableFuture的建立：

  • 使用new方法

    CompletableFuture<Double> futurePrice = new CompletableFuture<>();

  • 使用CompletableFuture#completedFuture靜態方法建立

    public static <U> CompletableFuture<U> completedFuture(U value) {
        return new CompletableFuture<U>((value == null) ? NIL : value);
    }

    參數的值爲任務執行完的結果，通常該方法在實際應用中較少應用

  • 使用 CompletableFuture#supplyAsync靜態方法建立
    supplyAsync有兩個重載方法：

    //方法一
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
        return asyncSupplyStage(asyncPool, supplier);
    }
    //方法二
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
                                                       Executor executor) {
        return asyncSupplyStage(screenExecutor(executor), supplier);
    }

  • 使用CompletableFuture#runAsync靜態方法建立

runAsync有兩個重載方法

//方法一
public static CompletableFuture<Void> runAsync(Runnable runnable) {
    return asyncRunStage(asyncPool, runnable);
}
//方法二
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) {
    return asyncRunStage(screenExecutor(executor), runnable);
}

說明：

• 兩個重載方法之間的區別 => 後者能夠傳入自定義Executor，前者是默認的，使用的ForkJoinPool
• supplyAsync和runAsync方法之間的區別 => 前者有返回值，後者無返回值
• Supplier是函數式接口，所以該方法須要傳入該接口的實現類，追蹤源碼會發如今run方法中會調用該接口的方法。所以使用該方法建立CompletableFuture對象只需重寫Supplier中的get方法，在get方法中定義任務便可。又由於函數式接口可使用Lambda表達式，和new建立CompletableFuture對象相比代碼會簡潔很多

• 結果的獲取：

對於結果的獲取CompltableFuture類提供了四種方式

//方式一
public T get()
//方式二
public T get(long timeout, TimeUnit unit)
//方式三
public T getNow(T valueIfAbsent)
//方式四
public T join()

說明：

• get()和get(long timeout, TimeUnit unit) => 在Future中就已經提供了，後者提供超時處理，若是在指定時間內未獲取結果將拋出超時異常
• getNow => 當即獲取結果不阻塞，結果計算已完成將返回結果或計算過程當中的異常，若是未計算完成將返回設定的valueIfAbsent值
• join => 方法裏不會拋出異常

示例：

public class AcquireResultTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //getNow方法測試
        CompletableFuture<String> cp1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(60 * 1000 * 60 );
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "hello world";
        });

        System.out.println(cp1.getNow("hello h2t"));

        //join方法測試
        CompletableFuture<Integer> cp2 = CompletableFuture.supplyAsync((()-> 1 / 0));
        System.out.println(cp2.join());

        //get方法測試
        CompletableFuture<Integer> cp3 = CompletableFuture.supplyAsync((()-> 1 / 0));
        System.out.println(cp3.get());
    }
}

說明：

• 第一個執行結果爲hello h2t，由於要先睡上1分鐘結果不能當即獲取
• join方法獲取結果方法裏不會拋異常，可是執行結果會拋異常，拋出的異常爲CompletionException
• get方法獲取結果方法裏將拋出異常，執行結果拋出的異常爲ExecutionException

• 異常處理：

使用靜態方法建立的CompletableFuture對象無需顯示處理異常，使用new建立的對象須要調用completeExceptionally方法設置捕獲到的異常，舉例說明：

CompletableFuture completableFuture = new CompletableFuture();
new Thread(() -> {
     try {
         //doSomething，調用complete方法將其餘方法的執行結果記錄在completableFuture對象中
         completableFuture.complete(null);
     } catch (Exception e) {
         //異常處理
         completableFuture.completeExceptionally(e);
      }
 }).start();

同步方法Pick異步方法查詢全部店鋪某個商品價格

店鋪爲一個列表：

private static List<Shop> shopList = Arrays.asList(
        new Shop("BestPrice"),
        new Shop("LetsSaveBig"),
        new Shop("MyFavoriteShop"),
        new Shop("BuyItAll")
);

同步方法：

private static List<String> findPriceSync(String product) {
    return shopList.stream()
            .map(shop -> String.format("%s price is %.2f",
                    shop.getName(), shop.getPrice(product)))  //格式轉換
            .collect(Collectors.toList());
}

異步方法：

private static List<String> findPriceAsync(String product) {
    List<CompletableFuture<String>> completableFutureList = shopList.stream()
            //轉異步執行
            .map(shop -> CompletableFuture.supplyAsync(
                    () -> String.format("%s price is %.2f",
                            shop.getName(), shop.getPrice(product))))  //格式轉換
            .collect(Collectors.toList());

    return completableFutureList.stream()
            .map(CompletableFuture::join)  //獲取結果不會拋出異常
            .collect(Collectors.toList());
}

性能測試結果：

Find Price Sync Done in 4141
Find Price Async Done in 1033

異步執行效率提升四倍

爲何仍須要CompletableFuture

在JDK1.8之前，經過調用線程池的submit方法可讓任務以異步的方式運行，該方法會返回一個Future對象，經過調用get方法獲取異步執行的結果：

private static List<String> findPriceFutureAsync(String product) {
    ExecutorService es = Executors.newCachedThreadPool();
    List<Future<String>> futureList = shopList.stream().map(shop -> es.submit(() -> String.format("%s price is %.2f",
            shop.getName(), shop.getPrice(product)))).collect(Collectors.toList());

    return futureList.stream()
            .map(f -> {
                String result = null;
                try {
                    result = f.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }

                return result;
            }).collect(Collectors.toList());
}

既生瑜何生亮，爲何仍須要引入CompletableFuture？
對於簡單的業務場景使用Future徹底沒有，可是想將多個異步任務的計算結果組合起來，後一個異步任務的計算結果須要前一個異步任務的值等等，使用Future提供的那點API就囊中羞澀，處理起來不夠優雅，這時候仍是讓CompletableFuture以聲明式的方式優雅的處理這些需求

其餘API介紹

whenComplete計算結果的處理：

對前面計算結果進行處理，沒法返回新值
提供了三個方法：

//方法一
public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
//方法二
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
//方法三
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)

說明：

• BiFunction<? super T,? super U,? extends V> fn參數 => 定義對結果的處理
• Executor executor參數 => 自定義線程池
• 以async結尾的方法將會在一個新的線程中執行組合操做

示例：

public class WhenCompleteTest {
    public static void main(String[] args) {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "hello");
        CompletableFuture<String> cf2 = cf1.whenComplete((v, e) ->
                System.out.println(String.format("value:%s, exception:%s", v, e)));
        System.out.println(cf2.join());
    }
}

thenApply轉換：

將前面計算結果的的CompletableFuture傳遞給thenApply，返回thenApply處理後的結果。能夠認爲經過thenApply方法實現CompletableFuture<T>至CompletableFuture<U>的轉換。白話一點就是將CompletableFuture的計算結果做爲thenApply方法的參數，返回thenApply方法處理後的結果
提供了三個方法：

//方法一
public <U> CompletableFuture<U> thenApply(
    Function<? super T,? extends U> fn) {
    return uniApplyStage(null, fn);
}

//方法二
public <U> CompletableFuture<U> thenApplyAsync(
    Function<? super T,? extends U> fn) {
    return uniApplyStage(asyncPool, fn);
}

//方法三
public <U> CompletableFuture<U> thenApplyAsync(
    Function<? super T,? extends U> fn, Executor executor) {
    return uniApplyStage(screenExecutor(executor), fn);
}

說明：

• Function<? super T,? extends U> fn參數 => 對前一個CompletableFuture 計算結果的轉化操做
• Executor executor參數 => 自定義線程池
• 以async結尾的方法將會在一個新的線程中執行組合操做

示例：

public class ThenApplyTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> result = CompletableFuture.supplyAsync(ThenApplyTest::randomInteger).thenApply((i) -> i * 8);
        System.out.println(result.get());
    }

    public static Integer randomInteger() {
        return 10;
    }
}

這裏將前一個CompletableFuture計算出來的結果擴大八倍

thenAccept結果處理：

thenApply也能夠歸類爲對結果的處理，thenAccept和thenApply的區別就是沒有返回值
提供了三個方法：

//方法一
public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
    return uniAcceptStage(null, action);
}

//方法二
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
    return uniAcceptStage(asyncPool, action);
}

//方法三
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
                                               Executor executor) {
    return uniAcceptStage(screenExecutor(executor), action);
}

說明：

• Consumer<? super T> action參數 => 對前一個CompletableFuture計算結果的操做
• Executor executor參數 => 自定義線程池
• 同理以async結尾的方法將會在一個新的線程中執行組合操做

示例：

public class ThenAcceptTest {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(ThenAcceptTest::getList).thenAccept(strList -> strList.stream()
                .forEach(m -> System.out.println(m)));
    }

    public static List<String> getList() {
        return Arrays.asList("a", "b", "c");
    }
}

將前一個CompletableFuture計算出來的結果打印出來

thenCompose異步結果流水化：

thenCompose方法能夠將兩個異步操做進行流水操做
提供了三個方法：

//方法一
public <U> CompletableFuture<U> thenCompose(
    Function<? super T, ? extends CompletionStage<U>> fn) {
    return uniComposeStage(null, fn);
}

//方法二
public <U> CompletableFuture<U> thenComposeAsync(
    Function<? super T, ? extends CompletionStage<U>> fn) {
    return uniComposeStage(asyncPool, fn);
}

//方法三
public <U> CompletableFuture<U> thenComposeAsync(
    Function<? super T, ? extends CompletionStage<U>> fn,
    Executor executor) {
    return uniComposeStage(screenExecutor(executor), fn);
}

說明：

• Function<? super T, ? extends CompletionStage<U>> fn參數 => 當前CompletableFuture計算結果的執行
• Executor executor參數 => 自定義線程池
• 同理以async結尾的方法將會在一個新的線程中執行組合操做

示例：

public class ThenComposeTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> result = CompletableFuture.supplyAsync(ThenComposeTest::getInteger)
                .thenCompose(i -> CompletableFuture.supplyAsync(() -> i * 10));
        System.out.println(result.get());
    }

    private static int getInteger() {
        return 666;
    }

    private static int expandValue(int num) {
        return num * 10;
    }
}

執行流程圖：

thenCombine組合結果：

thenCombine方法將兩個無關的CompletableFuture組合起來，第二個Completable並不依賴第一個Completable的結果
提供了三個方法：

//方法一
public <U,V> CompletableFuture<V> thenCombine( 
    CompletionStage<? extends U> other,
    BiFunction<? super T,? super U,? extends V> fn) {
    return biApplyStage(null, other, fn);
}
  //方法二
  public <U,V> CompletableFuture<V> thenCombineAsync(
      CompletionStage<? extends U> other,
      BiFunction<? super T,? super U,? extends V> fn) {
      return biApplyStage(asyncPool, other, fn);
  }

  //方法三
  public <U,V> CompletableFuture<V> thenCombineAsync(
      CompletionStage<? extends U> other,
      BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
      return biApplyStage(screenExecutor(executor), other, fn);
  }

說明：

• CompletionStage<? extends U> other參數 => 新的CompletableFuture的計算結果
• BiFunction<? super T,? super U,? extends V> fn參數 => 定義了兩個CompletableFuture對象完成計算後如何合併結果，該參數是一個函數式接口，所以可使用Lambda表達式
• Executor executor參數 => 自定義線程池
• 同理以async結尾的方法將會在一個新的線程中執行組合操做

示例：

public class ThenCombineTest {
    private static Random random = new Random();
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> result = CompletableFuture.supplyAsync(ThenCombineTest::randomInteger).thenCombine(
                CompletableFuture.supplyAsync(ThenCombineTest::randomInteger), (i, j) -> i * j
        );

        System.out.println(result.get());
    }

    public static Integer randomInteger() {
        return random.nextInt(100);
    }
}

將兩個線程計算出來的值作一個乘法在返回
執行流程圖：

allOf&anyOf組合多個CompletableFuture：

方法介紹：

//allOf
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
    return andTree(cfs, 0, cfs.length - 1);
}
//anyOf
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
    return orTree(cfs, 0, cfs.length - 1);
}

說明：

• allOf => 全部的CompletableFuture都執行完後執行計算。
• anyOf => 任意一個CompletableFuture執行完後就會執行計算

示例：

• allOf方法測試

  public class AllOfTest {
      public static void main(String[] args) throws ExecutionException, InterruptedException {
          CompletableFuture<Void> future1 = CompletableFuture.supplyAsync(() -> {
              System.out.println("hello");
              return null;
          });
          CompletableFuture<Void> future2 = CompletableFuture.supplyAsync(() -> {
              System.out.println("world"); return null;
          });
          CompletableFuture<Void> result = CompletableFuture.allOf(future1, future2);
          System.out.println(result.get());
      }
  }

  allOf方法沒有返回值，適合沒有返回值而且須要前面全部任務執行完畢才能執行後續任務的應用場景

• anyOf方法測試

  public class AnyOfTest {
      private static Random random = new Random();
      public static void main(String[] args) throws ExecutionException, InterruptedException {
          CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
              randomSleep();
              System.out.println("hello");
              return "hello";});
          CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
              randomSleep();
              System.out.println("world");
              return "world";
          });
          CompletableFuture<Object> result = CompletableFuture.anyOf(future1, future2);
          System.out.println(result.get());
     }
  
      private static void randomSleep() {
          try {
              Thread.sleep(random.nextInt(10));
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
      }
  }

  兩個線程都會將結果打印出來，可是get方法只會返回最早完成任務的結果。該方法比較適合只要有一個返回值就能夠繼續執行其餘任務的應用場景

注意點

不少方法都提供了異步實現【帶async後綴】，可是需當心謹慎使用這些異步方法，由於異步意味着存在上下文切換，可能性能不必定比同步好。若是須要使用異步的方法，先作測試，用測試數聽說話！！！

CompletableFuture的應用場景

存在IO密集型的任務能夠選擇CompletableFuture，IO部分交由另一個線程去執行。Logback、Log4j2異步日誌記錄的實現原理就是新起了一個線程去執行IO操做，這部分能夠以CompletableFuture.runAsync(()->{ioOperation();})的方式去調用，有關Logback異步日誌記錄的原理能夠參考這篇文章Logback異步日誌記錄

優化空間

supplyAsync執行任務底層實現：

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}
static <U> CompletableFuture<U> asyncSupplyStage(Executor e, Supplier<U> f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<U> d = new CompletableFuture<U>();
    e.execute(new AsyncSupply<U>(d, f));
    return d;
}

底層調用的是線程池去執行任務，而CompletableFuture中默認線程池爲ForkJoinPool

private static final Executor asyncPool = useCommonPool ?
        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

ForkJoinPool線程池的大小取決於CPU的核數。以前寫的爲何阿里巴巴要禁用Executors建立線程池？文章中說起過，CPU密集型任務線程池大小配置爲CPU核心數就能夠了，可是IO密集型，線程池的大小由CPU數量 CPU利用率 (1 + 線程等待時間/線程CPU時間)肯定。而CompletableFuture的應用場景就是IO密集型任務，所以默認的ForkJoinPool通常沒法達到最佳性能，咱們需本身根據業務建立線程池

最後附：示例代碼，歡迎Fork與Star

附往期文章：歡迎你的閱讀、點贊、評論

併發相關
1.爲何阿里巴巴要禁用Executors建立線程池？
2.本身的事情本身作，線程異常處理

設計模式相關：
1. 單例模式，你真的寫對了嗎？
2. (策略模式+工廠模式+map)套餐 Kill 項目中的switch case

JAVA8相關：
1. 使用Stream API優化代碼
2. 親，建議你使用LocalDateTime而不是Date哦

數據庫相關：
1. mysql數據庫時間類型datetime、bigint、timestamp的查詢效率比較
2. 很高興！終於踩到了慢查詢的坑

高效相關：
1. 擼一個Java腳手架，一統團隊項目結構風格

日誌相關：
1. 日誌框架，選擇Logback Or Log4j2？
2. Logback配置文件這麼寫，TPS提升10倍

工程相關：
1. 閒來無事，動手寫一個LRU本地緩存
2. Redis實現點贊功能模塊
3. JMX可視化監控線程池
4. 權限管理 【SpringSecurity篇】
5. Spring自定義註解從入門到精通
6. java模擬登錄優酷
7. QPS這麼高，那就來寫個多級緩存吧
8. java使用phantomjs進行截圖

其餘：
1. 使用try-with-resources優雅關閉資源
2. 老闆，用float存儲金額爲何要扣我工資