從CompletableFuture到異步編程設計

時間  2019-11-09
標籤 completablefuture 異步 編程 設計 简体版
原文   原文鏈接

從CompletableFuture到異步編程設計，筆者就分爲2部分來分享CompletableFuture異步編程設計，前半部分總結下CompletableFuture使用實踐，後半部分分享下CompletableFuture實現原理和異步編程設計機制。html

（ps：本文內容較多，請耐心閱讀。若是讀者瞭解CompletableFuture使用的話，能夠直接看後半部份內容；若是熟悉CompletableFuture及異步編程設計的話，能夠直接翻到文檔末尾點個「推薦」就行了，由於你已經掌握了Java異步設計精髓了 :) ，如有不正確地方，感謝評論區指正交流~ ）java

Java8新增了CompletableFuture類，該類實現了CompletionStage和Future接口，簡化了Java異步編程能力，該類方法較多，其實套路只有一個，那就是任務執行完成以後執行「回調」。git

CompletableFuture使用實踐

Java8新增的CompletableFuture 提供對異步計算的支持，能夠經過回調的方式處理計算結果。CompletableFuture 類實現了CompletionStage和Future接口，因此還能夠像以前使用Future那樣使用CompletableFuture ，儘管已再也不推薦這樣用了。github

CompletableFuture的建立編程

// 建立一個帶result的CompletableFuture
CompletableFuture<String> future = CompletableFuture.completedFuture("result");
future.get();
 
// 默認建立的CompletableFuture是沒有result的，這時調用future.get()會一直阻塞下去知道有result或者出現異常
future = new CompletableFuture<>();
try {
    future.get(1, TimeUnit.SECONDS);
} catch (Exception e) {
    // no care
}
 
// 給future填充一個result
future.complete("result");
assert "result".equals(future.get());
 
// 給future填充一個異常
future = new CompletableFuture<>();
future.completeExceptionally(new RuntimeException("exception"));
try {
    future.get();
} catch (Exception e) {
    assert "exception".equals(e.getCause().getMessage());
}

上面的示例是本身設置future的result，通常狀況下咱們都是讓其餘線程或者線程池來執行future這些異步任務。除了直接建立CompletableFuture 對象外（不推薦這樣使用），還可使用以下4個方法建立CompletableFuture 對象：api

// runAsync是Runnable任務，不帶返回值的，若是入參有executor，則使用executor來執行異步任務
public static CompletableFuture<Void>  runAsync(Runnable runnable)
public static CompletableFuture<Void>  runAsync(Runnable runnable, Executor executor)
// supplyAsync是待返回結果的異步任務
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier, Executor executor)

// 使用示例
CompletableFuture.runAsync(() -> {
    System.out.println("hello world");
}, executor);
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
});

若是入參不帶executor，則默認使用ForkJoinPool.commonPool()做爲執行異步任務的線程池；不然使用executor執行任務。oracle

CompletableFuture的完成動做app

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)
public CompletableFuture<T>     exceptionally(Function<Throwable,? extends T> fn)

// 使用示例
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
}).whenCompleteAsync((result, e) -> {
    System.out.println(result + " " + e);
}).exceptionally((e) -> {
    System.out.println("exception " + e);
    return "exception";
});

action是Action類型，從上面能夠看出它既能夠處理正常返回值也能夠處理異常，whenComplete會在任務執行完成後直接在當前線程內執行action動做，後綴帶Async的方法是交給其餘線程執行action（若是是線程池，執行action的可能和以前執行異步任務的是同一個線程），入參帶executor的交給executor線程池來執行action動做，當發生異常時，會在當前線程內執行exceptionally方法。框架

除了用上面的whenComplete來執行完成動做以外，還可使用handle方法，該方法能夠返回一個新的CompletableFuture的返回類型。異步

public <U> CompletableFuture<U>  handle(BiFunction<? super T,Throwable,? extends U> fn)
public <U> CompletableFuture<U>  handleAsync(BiFunction<? super T,Throwable,? extends U> fn)
public <U> CompletableFuture<U>  handleAsync(BiFunction<? super T,Throwable,? extends U> fn, Executor executor)

// handle方法示例：
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
});
CompletableFuture<Integer> f2 = f1.handle((r, e) -> {
    System.out.println("handle");
    return 1;
});

 除了使用handle方法來執行CompletableFuture返回類型轉換以外，還可使用thenApply方法，兩者不一樣的是前者會處理正常返回值和異常，所以能夠屏蔽異常，避免繼續拋出；然後者只能處理正常返回值，一旦有異常就會拋出。

public <U> CompletableFuture<U>  thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U>  thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U>  thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

// thenApply方法示例：
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
}).thenApply((r) -> {
    System.out.println(r);
    return "aaa";
}).thenApply((r) -> {
    System.out.println(r);
    return 1;
});

注意，上面的handle、thenApply都是返回新的CompletableFuture類型，若是隻是爲了在CompletableFuture完成以後執行某些消費動做，而不返回新的CompletableFuture類型，則可使用thenAccept方法。

public CompletableFuture<Void>  thenAccept(Consumer<? super T> action)
public CompletableFuture<Void>  thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void>  thenAcceptAsync(Consumer<? super T> action, Executor executor)

// thenAccept方法示例：
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
}).thenAccept(r -> {
    System.out.println(r);
}).thenAccept(r -> {
    // 這裏的r爲Void（null）了
    System.out.println(r);
});

上面的handle、thenApply和thenAppept都是對上一個CompletableFuture執行完的結果進行某些操做。那麼可不能夠同時對2個CompletableFuture執行結果執行某些操做呢？其實也是能夠的，使用thenAppeptBoth方法便可。注意，thenAppeptBoth和handle/thenApply/thenAppep的流程是同樣的，只不過thenAppeptBoth中包含了另外一個CompletableFuture對象（注意，這裏另外一個CompletableFuture對象的執行可並非上一個CompletableFuture執行結束纔開始執行的）。

public <U> CompletableFuture<Void>   thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
public <U> CompletableFuture<Void>   thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
public <U> CompletableFuture<Void>   thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action, Executor executor)
public     CompletableFuture<Void>  runAfterBoth(CompletionStage<?> other,  Runnable action)


// thenAcceptBoth方法示例：
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
}).thenAcceptBoth(CompletableFuture.completedFuture("result2"), (r1, r2) -> {
    System.out.println(r1 + "-" + r2);
});

注意，thenAcceptBoth方法是沒有返回值的（CompletableFuture<Void>），若是想用thenAcceptBoth這樣的功能而且還帶有返回值的CompletableFuture，那麼thenCombine方法就該上場了。

public <U,V> CompletableFuture<V>    thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V>    thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V>    thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

// thenCombine方法示例
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
}).thenCombine(CompletableFuture.completedFuture("result2"), (r1, r2) -> {
    System.out.println(r1 + "-" + r2);
    return r1 + "-" + r2;
});

thenAcceptBoth和runAfterBoth是當兩個CompletableFuture都計算完成，而下面的方法是當任意一個CompletableFuture計算完成的時候就會執行。

public CompletableFuture<Void>  acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void>  acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void>  acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)
 
public <U> CompletableFuture<U>  applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn)
public <U> CompletableFuture<U>  applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn)
public <U> CompletableFuture<U>  applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor)

若是當想在多個CompletableFuture都計算完成或者多個CompletableFuture中的一個計算完成後執行某個動做，可以使用方法 allOf 和 anyOf。

public static CompletableFuture<Void>      allOf(CompletableFuture<?>... cfs)
public static CompletableFuture<Object>    anyOf(CompletableFuture<?>... cfs)

若是當任務完成時並不想用CompletableFuture的結果，可使用thenRun方法來執行一個Runnable。

public CompletableFuture<Void>  thenRun(Runnable action)
public CompletableFuture<Void>  thenRunAsync(Runnable action)
public CompletableFuture<Void>  thenRunAsync(Runnable action, Executor executor)

以上方法都是在方法中返回一個值（或者不返回值），其實還能夠返回一個CompletableFuture，是否是很像類的組合同樣。

public <U> CompletableFuture<U>  thenCompose(Function<? super T,? extends CompletionStage<U>> fn)
public <U> CompletableFuture<U>  thenComposeAsync(Function<? super T,? extends CompletionStage<U>> fn)
public <U> CompletableFuture<U>  thenComposeAsync(Function<? super T,? extends CompletionStage<U>> fn, Executor executor)

// thenCompose方法示例：
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
}).thenCompose(r -> {
    System.out.println(r);
    return CompletableFuture.supplyAsync(() -> {
        System.out.println(r + " result2");
        return r + " result2";
    });
});

// 上面的代碼和下面的代碼效果是同樣的
CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world");
    return "result";
}).thenApply(r -> {
    System.out.println(r);
    return r;
}).thenApplyAsync(r -> {
    System.out.println(r + " result2");
    return r + " result2";
});

 CompletableFuture實現機制

先拋開 CompletableFuture 不談，若是程序中使用了線程池，如何才能在某個任務執行完成以後執行某些動做呢？其實Java線程池自己已經提供了任務執行先後的hook方法（beforeExecute和afterExecute），以下：

public class ThreadPoolExecutor extends AbstractExecutorService {
    // ...
    protected void beforeExecute(Thread t, Runnable r) { }
    protected void afterExecute(Runnable r, Throwable t) { }
    // ...
}

咱們只須要自定義線程池繼承ThreadPoolExecutor ，而後重寫beforeExecute和afterExecute方法便可，在afterExecute裏能夠執行一些動做。關於重寫ThreadPoolExecutor 的一個示例可點擊ListenableThreadPoolExecutor查看。

那麼CompletableFuture 的實現機制是怎樣的呢？其實，和上面的所說的「afterExecute機制」是相似的（本質是同樣的，回調機制），也是在任務執行完成後執行某些動做，以下代碼：

CompletableFuture.supplyAsync(() -> {
    // callable任務
    System.out.println("hello world");
    return "result";
}).thenApply(r -> {
    // 任務完成以後的動做（回調方法），相似於ThreadPoolExecutor.afterExecute方法
    System.out.println(r);
    return r;
});

上面的示例代碼其實主要完成了3個步驟，這3個步驟其實也是CompletableFuture的實現流程：

  1. 執行任務
  2. 添加任務完成以後的動做（回調方法）
  3. 執行回調

下面筆者就以上面的示例代碼，按照這3個步驟依次進行分析，此時建議讀者打開idea，寫個demo進行debug，這裏篇幅有限，筆者就只講解主要流程代碼，其餘代碼自行閱讀便可 :) 

一、執行任務

 執行任務的主要邏輯就是 AsyncSupply.run 方法：

public void run() {
    CompletableFuture<T> d; Supplier<T> f;
    // dep是當前CompletableFuture，fn是任務執行邏輯
    if ((d = dep) != null && (f = fn) != null) {
        dep = null; fn = null;
        if (d.result == null) {
            try {
                // 1 任務執行 & result cas設置
                d.completeValue(f.get());
            } catch (Throwable ex) {
                // 1.1 result cas異常設置
                d.completeThrowable(ex);
            }
        }
        // 2 任務完成，可能涉及到回調的執行
        d.postComplete();
    }
}

二、添加回調

添加回調方法的流程是從 thenApply 開始的：

public <U> CompletableFuture<U> thenApply(
    Function<? super T,? extends U> fn) {
    return uniApplyStage(null, fn);
}
private <V> CompletableFuture<V> uniApplyStage(
    Executor e, Function<? super T,? extends V> f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<V> d =  new CompletableFuture<V>();
    if (e != null || !d.uniApply(this, f, null)) {
        // 當上一個CompletableFuture未完成時，將該CompletableFuture添加
        // 到上一個CompletableFuture的statck中
        UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
        push(c);
        c.tryFire(SYNC);
    }
    return d;
}

CompletableFuture.statck 是 UniCompletion 類型的，該類型以下：

UniCompletion<T,V> {
    volatile Completion next;      // Treiber stack link
    Executor executor;                 // executor to use (null if none)
    CompletableFuture<V> dep;          // the dependent to complete
    CompletableFuture<T> src;          // source for action
}

3、執行回調

執行回調是從CompletableFuture.postComplete 開始的：

final void postComplete() {
    /*
     * On each step, variable f holds current dependents to pop
     * and run.  It is extended along only one path at a time,
     * pushing others to avoid unbounded recursion.
     */
    CompletableFuture<?> f = this; Completion h;
    while ((h = f.stack) != null ||
           (f != this && (h = (f = this).stack) != null)) {
        CompletableFuture<?> d; Completion t;
        // cas設置h.next到當前CompletableFuture.statck
        if (f.casStack(h, t = h.next)) {
            if (t != null) {
                if (f != this) {
                    pushStack(h);
                    continue;
                }
                h.next = null;    // detach
            }
            f = (d = h.tryFire(NESTED)) == null ? this : d;
        }
    }
}

// UniAccept
final CompletableFuture<Void> tryFire(int mode) {
    CompletableFuture<Void> d; CompletableFuture<T> a;
    if ((d = dep) == null ||
        !d.uniAccept(a = src, fn, mode > 0 ? null : this)) // 執行回調
        return null;
    dep = null; src = null; fn = null;
    // 返回當前CompletableFuture 或者 遞歸調用postComplete
    return d.postFire(a, mode);
}

看完上面3個步驟，是否是還不太清楚多個CompletableFuture之間的執行流程呢，說實話筆者第一次看的時候也是這樣的 :(，下面咱們換個例子並給出圖示來看：

CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
    System.out.println("hello world f1");
    sleep(1); // TimeUnit.SECONDS.sleep(1)
    return "result f1";
});
CompletableFuture<String> f2 = f1.thenApply(r -> {
    System.out.println(r);
    sleep(1);
    return "f2";
});
CompletableFuture<String> f3 = f2.thenApply(r -> {
    System.out.println(r);
    sleep(1);
    return "f2";
});

CompletableFuture<String> f4 = f1.thenApply(r -> {
    System.out.println(r);
    sleep(1);
    return "f2";
});
CompletableFuture<String> f5 = f4.thenApply(r -> {
    System.out.println(r);
    sleep(1);
    return "f2";
});
CompletableFuture<String> f6 = f5.thenApply(r -> {
    System.out.println(r);
    sleep(1);
    return "f2";
});

上面代碼對應的CompletableFuture及其Completion關係以下圖：

結合上圖和postComplete流程，能夠看出執行回調的順序是：f1 -> f4 -> f5 -> f6 -> f2 -> f3。（若是這裏沒看懂，能夠回過頭再看下postComplete方法的源碼~）

異步編程設計

分析完了CompletableFuture，相信你們都已經對CompletableFuture的設計與實現有了進一步的理解。那麼對於異步編程有哪些實際應用場景，其本質究竟是什麼呢？

異步處理的本質其實就是回調（系統層藉助於指針來實現，準確來講是函數指針），用戶提供一個回調方法，回調函數不是由該函數的實現方直接調用，而是在特定的事件或條件發生時由另外的一方調用的，用於對該事件或條件進行響應。從「宏觀」來看，CompletableFuture的實現其實很簡單，就是回調，即在任務執行完成以後進行回調，回調中可能涉及到其餘操做，好比下一個回調或者執行下一個任務。

異步編程在應用場景較多，不少語言，好比Node.js，採用回調的方式實現異步編程。Java的一些框架，好比Netty，本身擴展了Java的 Future接口，提供了addListener等多個擴展方法：

ServerBootstrap boot = new ServerBootstrap();
boot.group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .localAddress(8080)
    .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new EchoHandler());
        }
    });

dubbo中consumer對於RPC response的處理是基於回調機制的，Google guava也提供了通用的擴展Future:ListenableFuture、SettableFuture 以及輔助類Futures等,方便異步編程。

final String name = ...;
inFlight.add(name);
ListenableFuture<Result> future = service.query(name);
future.addListener(new Runnable() {
  public void run() {
    processedCount.incrementAndGet();
    inFlight.remove(name);
    lastProcessed.set(name);
    logger.info("Done with {0}", name);
  }
}, executor);

 

參考資料：

一、Java CompletableFuture 詳解

二、https://www.cnblogs.com/aniao/p/aniao_cf.html

三、https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程