Guava 的沖擊

由于 JDK1.5 Futrure 的 get 方法獲取任務結果必須阻塞等待，Google 看不下去了，開發了 Guava 庫

 public static void main(String[] args) throws Exception {
        // 裝飾器模式
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));

        ListenableFuture<String> future = service.submit(new MyCallable());

        // 異步非阻塞：觀察者模式  當ListenableFuture完成時，需要執行的程式
        Futures.addCallback(future, new FutureCallback<String>() {
            @Override
            public void onSuccess(@Nullable String s) {

            }

            @Override
            public void onFailure(Throwable throwable) {
                throwable.printStackTrace();
            }
        }, service);

        log.info("do something in main");
        Thread.sleep(1000);
        String result = future.get();
        log.info("result：{}", result);
    }

于是 java 開始慌了，jdk8 趕緊出招，于是直接抄個類似的即可，

用多執行緒優化性能，其實不過就是將串行操作變成并行操作，在串行轉換成并行的程序中，一定會涉及到異步化，如下

// 以下兩個方法都是耗時操作
doBizA();
doBizB();

現在是串行的，

為了提升性能，得把它們并行化，那具體實施起來該怎么做呢？
如下，創建倆子執行緒去執行即可，

new Thread(()->doBizA()).start(); 
new Thread(()->doBizB()).start();

主執行緒無需等待doBizA、doBizB的執行結果，即doBizA()和doBizB()兩個操作是異步的，

異步化，是并行方案得以實施的基礎，利用多執行緒優化性能這個核心方案得以實施的基礎，所以異步編程很重要，因為優化性能是大廠的核心需求，JDK8提供CompletableFuture支持異步編程，

CompletableFuture的核心優勢

3個任務：

任務1：洗水壺、燒開水
任務2負責洗茶壺、洗茶杯和拿茶葉
任務3負責泡茶
任務3要等待任務1和任務2都完成后才能開始，

下面是代碼實作會發現：無需手工維護執行緒，無需關注給任務分配執行緒

//任務1：洗水壺->燒開水
CompletableFuture<Void> f1 = 
  CompletableFuture.runAsync(()->{
  System.out.println("T1:洗水壺...");
  sleep(1, TimeUnit.SECONDS);

  System.out.println("T1:燒開水...");
  sleep(15, TimeUnit.SECONDS);
});
//任務2：洗茶壺->洗茶杯->拿茶葉
CompletableFuture<String> f2 = 
  CompletableFuture.supplyAsync(()->{
  System.out.println("T2:洗茶壺...");
  sleep(1, TimeUnit.SECONDS);

  System.out.println("T2:洗茶杯...");
  sleep(2, TimeUnit.SECONDS);

  System.out.println("T2:拿茶葉...");
  sleep(1, TimeUnit.SECONDS);
  return "龍井";
});

//任務3：任務1和任務2完成后執行：泡茶
CompletableFuture<String> f3 = 
  f1.thenCombine(f2, (__, tf)->{
    System.out.println("T1:拿到茶葉:" + tf);
    System.out.println("T1:泡茶...");
    return "上茶:" + tf;
  });
//等待任務3執行結果
System.out.println(f3.join());

void sleep(int t, TimeUnit u) {
  try {
    u.sleep(t);
  }catch(InterruptedException e){}
}

// 一次執行結果：
T1:洗水壺...
T2:洗茶壺...
T1:燒開水...
T2:洗茶杯...
T2:拿茶葉...
T1:拿到茶葉:龍井
T1:泡茶...
上茶:龍井

創建CompletableFuture物件

創建CompletableFuture物件主要靠下面代碼中展示的這4個

靜態方法

頭兩個使用默認執行緒池，

runAsync(Runnable runnable)

Runnable 介面的run()方法沒有回傳值

supplyAsync(Supplier supplier)

Supplier介面的get()方法有回傳值

CompletableFuture默認使用公共的ForkJoinPool執行緒池，這個執行緒池默認創建的執行緒數是CPU的核數（也可以通過JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism設定ForkJoinPool執行緒池的執行緒數），

若所有CompletableFuture共享一個執行緒池，那么一旦有任務執行一些很慢的I/O操作，就會導致執行緒池中所有執行緒都阻塞在I/O操作上，從而造成執行緒饑餓，進而影響整個系統的性能，所以，推薦根據不同業務型別創建不同的執行緒池，以避免互相干擾，下兩個方法可指定執行緒池

runAsync(Runnable runnable, Executor executor)

supplyAsync(Supplier supplier, Executor executor)

創建完CompletableFuture物件后，會自動異步執行runnable.run()或者supplier.get()，對一個異步操作，我們關注：

異步操作什么時候結束
如何獲取異步操作的執行結果

因為CompletableFuture類實作了Future介面，所以這些都是通過Future介面解決的，
CompletableFuture類還實作了CompletionStage介面

CompletionStage介面

任務有時序關系，比如

串行
比如燒水泡茶，其中洗水壺和燒開水
并行
洗水壺、燒開水和洗茶壺、洗茶杯這兩組任務之間
匯聚
燒開水、拿茶葉這倆任務和泡茶就是匯聚

CompletionStage介面可清晰描述任務之間的這種時序關系，例如

f3 = f1.thenCombine(f2, ()->{})

描述的就是一種匯聚關系，燒水泡茶中的匯聚關系是一種

AND 聚合關系
AND指所有依賴的任務（燒開水和拿茶葉）都完成后才開始執行當前任務（泡茶）

還有

OR聚合關系
OR指的是依賴的任務只要有一個完成就可以執行當前任務，

1 串行關系

CompletionStage介面里面描述串行關系，主要是thenApply、thenAccept、thenRun和thenCompose這四個系列的介面，

thenApply系

fn的型別是介面Function<T, R>，這個介面里與CompletionStage相關的方法是 R apply(T t)

該方法既能接收引數也支持回傳值，所以thenApply系列方法回傳的是CompletionStage<U>，

thenAccept系

引數consumer的型別是介面Consumer<T>，這個介面里與CompletionStage相關的方法是
void accept(T t)

該方法雖然支持引數，但不支持回傳值，所以thenAccept系方法回傳值是CompletionStage，

thenRun系

引數是Runnable，所以action既不能接收引數也不支持回傳值，所以thenRun系列方法回傳的也是CompletionStage<Void>，

Async表示異步執行fn、consumer或action，

thenCompose系

這個系列的方法會新創建出一個子流程，最終結果和thenApply系相同，

看如何使用thenApply()，
supplyAsync()啟動一個異步流程，之后是兩個串行操作，雖然這是一個異步流程，但任務1、2、3是串行執行，即2依賴1的執行結果，3依賴2的執行結果，

2 AND匯聚

主要是thenCombine、thenAcceptBoth和runAfterBoth系介面

3 OR匯聚

主要是applyToEither、acceptEither和runAfterEither系介面

CompletionStage applyToEither(other, fn);
CompletionStage applyToEitherAsync(other, fn);
CompletionStage acceptEither(other, consumer);
CompletionStage acceptEitherAsync(other, consumer);
CompletionStage runAfterEither(other, action);
CompletionStage runAfterEitherAsync(other, action);

如何使用applyToEither()描述OR匯聚關系，

CompletableFuture<String> f1 = 
  CompletableFuture.supplyAsync(()->{
    int t = getRandom(5, 10);
    sleep(t, TimeUnit.SECONDS);
    return String.valueOf(t);
});

CompletableFuture<String> f2 = 
  CompletableFuture.supplyAsync(()->{
    int t = getRandom(5, 10);
    sleep(t, TimeUnit.SECONDS);
    return String.valueOf(t);
});

CompletableFuture<String> f3 = 
  f1.applyToEither(f2,s -> s);

System.out.println(f3.join());

CompletableFuture 中各種關系（并行、串行、聚合），支持的各種場景，比如：執行緒A 等待執行緒B或執行緒C等待執行緒A、B ，

其實CountdownLatch、ThreadPoolExecutor 和Future 就是來解決這些關系場景的，現在有了 completableFuture，可以優先考慮使用 CompletableFuture，

4 例外處理

fn、consumer、action的核心方法都不允許拋受檢例外，但無法限制它們拋運行時例外，例如下面的代碼，執行 1/0 就會出現除0錯誤的運行時例外，
非異步編程里，可以用try/catch捕獲并處理例外，異步編程里該如何處理呢？

CompletionStage給出的方案很簡單，使用這些方法處理例外和串行操作一樣的，而且還支持鏈式編程，

CompletionStage exceptionally(fn);
CompletionStage<R> whenComplete(consumer);
CompletionStage<R> whenCompleteAsync(consumer);
CompletionStage<R> handle(fn);
CompletionStage<R> handleAsync(fn);

exceptionally()類似try/catch中的catch
whenComplete()和handle()類似try/finally的finally，無論是否發生例外都會執行whenComplete()中的回呼方法consumer和handle()中的回呼方法fn
whenComplete()不支持回傳結果，handle()支持回傳結果，

學了這么多，最后來看個例子：

//采購訂單
PurchersOrder po;
CompletableFuture<Boolean> cf = 
  CompletableFuture.supplyAsync(()->{
    // 在MySQL中查詢規則
    return findRuleByJdbc();
  }).thenApply(r -> {
    // 規則校驗
    return check(po, r);
});
Boolean isOk = cf.join();

如上代碼問題在于：

讀資料庫屬于I/O操作，應定制單獨的執行緒池，避免執行緒饑餓
查出來的結果做為下一步處理的條件，若結果為空，沒有對應處理
例外未處理

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/278074.html

標籤：java

上一篇：Java基礎語法（六）——陣列的定義與使用

下一篇：二、Java之萬年歷

手把手教學妹CompletableFuture異步化,性能關系直接起飛!