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

寫在前面

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

• CompletableFuture的使用
• CompletableFure異步和同步的性能測試
• 已經有了Future為什麼仍需要在JDK1.8中引入CompletableFuture
• CompletableFuture的應用場景
• 對CompletableFuture的使用優化

場景說明

查詢所有商店某個商品的價格並返回，並且查詢商店某個商品的價格的API為同步 一個Shop類，提供一個名為getPrice的同步方法

• 店鋪類：Shop.java

<code>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();
        }
    }
}/<code>

查詢商品的價格為同步方法，並通過sleep方法模擬其他操作。這個場景模擬了當需要調用第三方API，但第三方提供的是同步API，在無法修改第三方API時如何設計代碼調用提高應用的性能和吞吐量，這時候可以使用CompletableFuture類

CompletableFuture使用

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

• CompletableFuture的創建：

使用new方法

<code>CompletableFuture<double> futurePrice = new CompletableFuture<>(); /<double>/<code>

使用CompletableFuture#completedFuture靜態方法創建

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

參數的值為任務執行完的結果，一般該方法在實際應用中較少應用

//方法一

<code>public static  CompletableFuture supplyAsync(Supplier supplier) {
  return asyncSupplyStage(asyncPool, supplier);
 }/<code>

//方法二

<code>public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor) { return asyncSupplyStage(screenExecutor(executor), supplier); }/<code>

<code> //方法一 /<code>

說明：

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

• 結果的獲取：

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

<code>//方式一
public T get()

//方式二

public T get(long timeout, TimeUnit unit)

//方式三 


public T getNow(T valueIfAbsent)

//方式四

public T join()/<code>

說明：

示例：

<code>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());

    }

}/<integer>/<integer>/<string>/<code>

說明：

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

<code>CompletableFuture completableFuture = new CompletableFuture();
new Thread(() -> {

     try {

         //doSomething，調用complete方法將其他方法的執行結果記錄在completableFuture對象中 


         completableFuture.complete(null);

     } catch (Exception e) {

         //異常處理

         completableFuture.completeExceptionally(e);

      }

 }).start();/<code>

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

店鋪為一個列表：

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

同步方法：

<code>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());
}/<string>/<code>

異步方法：

<code>private static List<string> findPriceAsync(String product) {
    List<completablefuture>> 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());
}/<completablefuture>/<string>/<code>

性能測試結果：

<code>Find Price Sync Done in 4141
Find Price Async Done in 1033/<code>

異步執行效率提高四倍

為什麼仍需要CompletableFuture

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

<code>private static List<string> findPriceFutureAsync(String product) {
    ExecutorService es = Executors.newCachedThreadPool();
    List<future>> 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()); 

}/<future>/<string>/<code>

既生瑜何生亮，為什麼仍需要引入CompletableFuture？
對於簡單的業務場景使用Future完全沒有，但是想將多個異步任務的計算結果組合起來，後一個異步任務的計算結果需要前一個異步任務的值等等，使用Future提供的那點API就囊中羞澀，處理起來不夠優雅，這時候還是讓CompletableFuture以聲明式的方式優雅的處理這些需求。而且在Future編程中想要拿到Future的值然後拿這個值去做後續的計算任務，只能通過輪詢的方式去判斷任務是否完成這樣非常佔CPU並且代碼也不優雅，用偽代碼表示如下：

<code>while(future.isDone()) {
    result = future.get();
    doSomrthingWithResult(result);
} /<code>

但CompletableFuture提供了API幫助我們實現這樣的需求

其他API介紹

whenComplete計算結果的處理：

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

<code>//方法一
public CompletableFuture whenComplete(BiConsumer super T,? super Throwable> action) 

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

說明：

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

示例：

<code>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());
    }
}/<string>/<string>/<code>

thenApply轉換：

將前面計算結果的的CompletableFuture傳遞給thenApply，返回thenApply處理後的結果。可以認為通過thenApply方法實現CompletableFuture至CompletableFuture的轉換。白話一點就是將CompletableFuture的計算結果作為thenApply方法的參數，返回thenApply方法處理後的結果

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

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

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

說明：

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

<code>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;
    }
}/<integer>/<code>

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

thenAccept結果處理：

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

<code>//方法一
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);
}/<void>/<void>/<void>/<code>

說明：

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

<code>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");
    }
}/<string>/<code>

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

thenCompose異步結果流水化：

thenCompose方法可以將兩個異步操作進行流水操作
提供了三個方法：

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

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

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

說明：

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

<code>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; 

    }
}/<integer>/<code>

執行流程圖：

thenCombine組合結果：

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

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

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

說明：

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

示例：

<code>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);
    }
}/<integer>/<code>

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

allOf&anyOf組合多個CompletableFuture：

方法介紹：

<code>//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);
}/<object>/<void>/<code>

說明：

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

示例：

• allOf方法測試

<code>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());

    }

}/<void>/<void>/<void>/<code>

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

• anyOf方法測試

<code>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();

        }

    }/<object>/<string>/<string>/<code>

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

注意點

很多方法都提供了異步實現【帶async後綴】，但是需小心謹慎使用這些異步方法，因為異步意味著存在上下文切換，可能性能不一定比同步好。如果需要使用異步的方法，先做測試，用測試數據說話！！！

CompletableFuture的應用場景

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

優化空間

supplyAsync執行任務底層實現：

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

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

<code>private static final Executor asyncPool = useCommonPool ?
/<code>

ForkJoinPool線程池的大小取決於CPU的核數。之前寫的為什麼阿里巴巴要禁用Executors創建線程池？文章中提及過，CPU密集型任務線程池大小配置為CPU核心數就可以了，但是IO密集型，線程池的大小由**CPU數量 * CPU利用率 * (1 + 線程等待時間/線程CPU時間)**確定。而CompletableFuture的應用場景就是IO密集型任務，因此默認的ForkJoinPool一般無法達到最佳性能，我們需自己根據業務創建線程池

作者：何甜甜在嗎
原文鏈接：https://juejin.im/post/5dfb5bc951882512420b06c5

閱讀更多 追逐仰望星空 的文章

關鍵字: 免受 異步 Docker