Java8新的異步編程方式 CompletableFuture(一)

一. Future

JDK 5引入了Future模式。Future接口是Java多線程Future模式的實現，在java.util.concurrent包中，能夠來進行異步計算。

Future模式是多線程設計經常使用的一種設計模式。Future模式能夠理解成：我有一個任務，提交給了Future，Future替我完成這個任務。期間我本身能夠去作任何想作的事情。一段時間以後，我就即可以從Future那兒取出結果。

Future的接口很簡單，只有五個方法。

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}複製代碼

Future接口的方法介紹以下：

• boolean cancel (boolean mayInterruptIfRunning) 取消任務的執行。參數指定是否當即中斷任務執行，或者等等任務結束
• boolean isCancelled () 任務是否已經取消，任務正常完成前將其取消，則返回 true
• boolean isDone () 任務是否已經完成。須要注意的是若是任務正常終止、異常或取消，都將返回true
• V get () throws InterruptedException, ExecutionException 等待任務執行結束，而後得到V類型的結果。InterruptedException 線程被中斷異常， ExecutionException任務執行異常，若是任務被取消，還會拋出CancellationException
• V get (long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException 同上面的get功能同樣，多了設置超時時間。參數timeout指定超時時間，uint指定時間的單位，在枚舉類TimeUnit中有相關的定義。若是計 算超時，將拋出TimeoutException

通常狀況下，咱們會結合Callable和Future一塊兒使用，經過ExecutorService的submit方法執行Callable，並返回Future。

ExecutorService executor = Executors.newCachedThreadPool();

        Future<String> future = executor.submit(() -> { //Lambda 是一個 callable， 提交後便當即執行，這裏返回的是 FutureTask 實例
            System.out.println("running task");
            Thread.sleep(10000);
            return "return task";
        });

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }

        System.out.println("do something else");  //前面的的 Callable 在其餘線程中運行着，能夠作一些其餘的事情

        try {
            System.out.println(future.get());  //等待 future 的執行結果，執行完畢以後打印出來
        } catch (InterruptedException e) {
        } catch (ExecutionException e) {

        } finally {
            executor.shutdown();
        }複製代碼

比起future.get()，其實更推薦使用get (long timeout, TimeUnit unit) 方法，設置了超時時間能夠防止程序無限制的等待future的結果。

二. CompletableFuture介紹

2.1 Future模式的缺點

• Future雖然能夠實現獲取異步執行結果的需求，可是它沒有提供通知的機制，咱們沒法得知Future何時完成。

• 要麼使用阻塞，在future.get()的地方等待future返回的結果，這時又變成同步操做。要麼使用isDone()輪詢地判斷Future是否完成，這樣會耗費CPU的資源。

2.2 CompletableFuture介紹

Netty、Guava分別擴展了Java 的 Future 接口，方便異步編程。

Java 8新增的CompletableFuture類正是吸取了全部Google Guava中ListenableFuture和SettableFuture的特徵，還提供了其它強大的功能，讓Java擁有了完整的非阻塞編程模型：Future、Promise 和 Callback(在Java8以前，只有無Callback 的Future)。

CompletableFuture可以將回調放到與任務不一樣的線程中執行，也能將回調做爲繼續執行的同步函數，在與任務相同的線程中執行。它避免了傳統回調最大的問題，那就是可以將控制流分離到不一樣的事件處理器中。

CompletableFuture彌補了Future模式的缺點。在異步的任務完成後，須要用其結果繼續操做時，無需等待。能夠直接經過thenAccept、thenApply、thenCompose等方式將前面異步處理的結果交給另一個異步事件處理線程來處理。

三. CompletableFuture特性

3.1 CompletableFuture的靜態工廠方法

方法名	描述
runAsync(Runnable runnable)	使用ForkJoinPool.commonPool()做爲它的線程池執行異步代碼。
runAsync(Runnable runnable, Executor executor)	使用指定的thread pool執行異步代碼。
supplyAsync(Supplier<U> supplier)	使用ForkJoinPool.commonPool()做爲它的線程池執行異步代碼，異步操做有返回值
supplyAsync(Supplier<U> supplier, Executor executor)	使用指定的thread pool執行異步代碼，異步操做有返回值

runAsync 和 supplyAsync 方法的區別是runAsync返回的CompletableFuture是沒有返回值的。

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println("Hello");
        });

        try {
            future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        System.out.println("CompletableFuture");複製代碼

而supplyAsync返回的CompletableFuture是由返回值的，下面的代碼打印了future的返回值。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        System.out.println("CompletableFuture");複製代碼

3.2 Completable

方法名	描述
complete(T t)	完成異步執行，並返回future的結果
completeExceptionally(Throwable ex)	異步執行不正常的結束

future.get()在等待執行結果時，程序會一直block，若是此時調用complete(T t)會當即執行。

CompletableFuture<String> future  = CompletableFuture.supplyAsync(() -> "Hello");

        future.complete("World");

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }複製代碼

執行結果：

World複製代碼

能夠看到future調用complete(T t)會當即執行。可是complete(T t)只能調用一次，後續的重複調用會失效。

若是future已經執行完畢可以返回結果，此時再調用complete(T t)則會無效。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        future.complete("World");

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }複製代碼

執行結果：

Hello複製代碼

若是使用completeExceptionally(Throwable ex)則拋出一個異常，而不是一個成功的結果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        future.completeExceptionally(new Exception());

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }複製代碼

執行結果：

java.util.concurrent.ExecutionException: java.lang.Exception
...複製代碼