Java CompletableFuture 詳解

yangtaochina 8年前發布 | 26K 次閱讀 Java Java開發

來自： http://colobu.com/2016/02/29/Java-CompletableFuture/

Future 是Java 5添加的類，用來描述一個異步計算的結果。你可以使用 isDone 方法檢查計算是否完成，或者使用 get 阻塞住調用線程，直到計算完成返回結果，你也可以使用 cancel 方法停止任務的執行。

publicclassBasicFuture{
publicstaticvoidmain(String[] args)throwsExecutionException, InterruptedException {
 ExecutorService es = Executors.newFixedThreadPool(10);
 Future<Integer> f = es.submit(() ->{
// 長時間的異步計算
// ……
// 然后返回結果
return100;
 });
// while(!f.isDone())
// ;
 f.get();
 }
}

雖然 Future 以及相關使用方法提供了異步執行任務的能力，但是對于結果的獲取卻是很不方便，只能通過阻塞或者輪詢的方式得到任務的結果。阻塞的方式顯然和我們的異步編程的初衷相違背，輪詢的方式又會耗費無謂的CPU資源，而且也不能及時地得到計算結果，為什么不能用觀察者設計模式當計算結果完成及時通知監聽者呢？

很多語言，比如Node.js，采用回調的方式實現異步編程。Java的一些框架，比如Netty，自己擴展了Java的 Future 接口，提供了 addListener 等多個擴展方法：

ChannelFuture future = bootstrap.connect(newInetSocketAddress(host, port));
 future.addListener(newChannelFutureListener()
 {
@Override
publicvoidoperationComplete(ChannelFuture future)throwsException
 {
if(future.isSuccess()) {
// SUCCESS
 }
else{
// FAILURE
 }
 }
 });

Google guava也提供了通用的擴展Future: ListenableFuture 、 SettableFuture 以及輔助類 Futures 等,方便異步編程。

finalString name = ...;
inFlight.add(name);
ListenableFuture<Result> future = service.query(name);
future.addListener(newRunnable() {
publicvoidrun() {
 processedCount.incrementAndGet();
 inFlight.remove(name);
 lastProcessed.set(name);
 logger.info("Done with {0}", name);
 }
}, executor);

Scala也提供了簡單易用且功能強大的Future/Promise 異步編程模式。

作為正統的Java類庫，是不是應該做點什么，加強一下自身庫的功能呢？

在Java 8中, 新增加了一個包含50個方法左右的類: CompletableFuture ，提供了非常強大的Future的擴展功能，可以幫助我們簡化異步編程的復雜性，提供了函數式編程的能力，可以通過回調的方式處理計算結果，并且提供了轉換和組合CompletableFuture的方法。

下面我們就看一看它的功能吧。

主動完成計算

CompletableFuture類實現了 CompletionStage 和 Future 接口，所以你還是可以像以前一樣通過阻塞或者輪詢的方式獲得結果，盡管這種方式不推薦使用。

publicTget()
publicTget(longtimeout, TimeUnit unit)
publicTgetNow(T valueIfAbsent)
publicTjoin()

getNow 有點特殊，如果結果已經計算完則返回結果或者拋出異常，否則返回給定的 valueIfAbsent 值。

join 返回計算的結果或者拋出一個unchecked異常(CompletionException)，它和 get 對拋出的異常的處理有些細微的區別，你可以運行下面的代碼進行比較：

</div>

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
inti =1/0;
return100;
});
//future.join();
future.get();

盡管Future可以代表在另外的線程中執行的一段異步代碼，但是你還是可以在本身線程中執行：

publicstaticCompletableFuture<Integer>compute() {
finalCompletableFuture<Integer> future =newCompletableFuture<>();
returnfuture;
}</pre> 
 
 上面的代碼中 future 沒有關聯任何的 Callback 、線程池、異步任務等，如果客戶端調用 future.get 就會一致傻等下去。你可以通過下面的代碼完成一個計算，觸發客戶端的等待： 

f.complete(100); 
 當然你也可以拋出一個異常，而不是一個成功的計算結果：

f.completeExceptionally(newException()); 
 完整的代碼如下：

publicclassBasicMain{
publicstaticCompletableFuture<Integer>compute() {
finalCompletableFuture<Integer> future =newCompletableFuture<>();
returnfuture;
 }
publicstaticvoidmain(String[] args)throwsException {
finalCompletableFuture<Integer> f = compute();
 class Client extends Thread {
 CompletableFuture<Integer> f;
 Client(String threadName, CompletableFuture<Integer> f) {
super(threadName);
this.f = f;
 }
@Override
publicvoidrun() {
try{
 System.out.println(this.getName() +": "+ f.get());
 } catch(InterruptedException e) {
 e.printStackTrace();
 } catch(ExecutionException e) {
 e.printStackTrace();
 }
 }
 }
newClient("Client1", f).start();
newClient("Client2", f).start();
 System.out.println("waiting");
 f.complete(100);
//f.completeExceptionally(new Exception());
 System.in.read();
 }
} 
  可以看到我們并沒有把 f.complete(100); 放在另外的線程中去執行，但是在大部分情況下我們可能會用一個線程池去執行這些異步任務。 CompletableFuture.complete() 、 CompletableFuture.completeExceptionally 只能被調用一次。但是我們有兩個后門方法可以重設這個值: obtrudeValue 、 obtrudeException ，但是使用的時候要小心，因為 complete 已經觸發了客戶端，有可能導致客戶端會得到不期望的結果。 

創建CompletableFuture對象。
 CompletableFuture.completedFuture 是一個靜態輔助方法，用來返回一個已經計算好的 CompletableFuture 。 
publicstatic<U> CompletableFuture<U>completedFuture(U value) 
  而以下四個靜態方法用來為一段異步執行的代碼創建 CompletableFuture 對象： 

publicstaticCompletableFuture<Void>runAsync(Runnable runnable)
publicstaticCompletableFuture<Void>runAsync(Runnable runnable, Executor executor)
publicstatic<U> CompletableFuture<U>supplyAsync(Supplier<U> supplier)
publicstatic<U> CompletableFuture<U>supplyAsync(Supplier<U> supplier, Executor executor) 
  以 Async 結尾并且沒有指定 Executor 的方法會使用 ForkJoinPool.commonPool() 作為它的線程池執行異步代碼。 

 
   runAsync 方法也好理解，它以 Runnable 函數式接口類型為參數，所以 CompletableFuture 的計算結果為空。 

   supplyAsync 方法以 Supplier<U> 函數式接口類型為參數, CompletableFuture 的計算結果類型為 U 。 

 
因為方法的參數類型都是函數式接口，所以可以使用lambda表達式實現異步任務，比如：
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
//長時間的計算任務
return"·00";
}); 
 計算結果完成時的處理

 當 CompletableFuture 的計算結果完成，或者拋出異常的時候，我們可以執行特定的 Action 。主要是下面的方法： 
publicCompletableFuture<T>whenComplete(BiConsumer<?superT,?superThrowable> action)
publicCompletableFuture<T>whenCompleteAsync(BiConsumer<?superT,?superThrowable> action)
publicCompletableFuture<T>whenCompleteAsync(BiConsumer<?superT,?superThrowable> action, Executor executor)
publicCompletableFuture<T>exceptionally(Function<Throwable,? extends T> fn) 
  
   可以看到 Action 的類型是 BiConsumer<? super T,? super Throwable> ，它可以處理正常的計算結果，或者異常情況。 

 方法不以 Async 結尾，意味著 Action 使用相同的線程執行，而 Async 可能會使用其它的線程去執行(如果使用相同的線程池，也可能會被同一個線程選中執行)。 
</div>
 注意這幾個方法都會返回 CompletableFuture ，當 Action 執行完畢后它的結果返回原始的 CompletableFuture 的計算結果或者返回異常。 
publicclassMain{
privatestaticRandom rand =newRandom();
privatestaticlongt = System.currentTimeMillis();
staticintgetMoreData() {
 System.out.println("begin to start compute");
try{
 Thread.sleep(10000);
 } catch(InterruptedException e) {
thrownewRuntimeException(e);
 }
 System.out.println("end to start compute. passed "+ (System.currentTimeMillis() - t)/1000+" seconds");
returnrand.nextInt(1000);
 }
publicstaticvoidmain(String[] args)throwsException {
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(Main::getMoreData);
 Future<Integer> f = future.whenComplete((v, e) -> {
 System.out.println(v);
 System.out.println(e);
 });
 System.out.println(f.get());
 System.in.read();
 }
} 
  exceptionally 方法返回一個新的CompletableFuture，當原始的CompletableFuture拋出異常的時候，就會觸發這個CompletableFuture的計算，調用function計算值，否則如果原始的CompletableFuture正常計算完后，這個新的CompletableFuture也計算完成，它的值和原始的CompletableFuture的計算的值相同。也就是這個 exceptionally 方法用來處理異常的情況。 

 下面一組方法雖然也返回CompletableFuture對象，但是對象的值和原來的CompletableFuture計算的值不同。當原先的CompletableFuture的值計算完成或者拋出異常的時候，會觸發這個CompletableFuture對象的計算，結果由 BiFunction 參數計算而得。因此這組方法兼有 whenComplete 和轉換的兩個功能。 
public<U> CompletableFuture<U>handle(BiFunction<?superT,Throwable,? extends U> fn)
public<U> CompletableFuture<U>handleAsync(BiFunction<?superT,Throwable,? extends U> fn)
public<U> CompletableFuture<U>handleAsync(BiFunction<?superT,Throwable,? extends U> fn, Executor executor) 
  同樣，不以 Async 結尾的方法由原來的線程計算，以 Async 結尾的方法由默認的線程池 ForkJoinPool.commonPool() 或者指定的線程池 executor 運行。 

轉換
 CompletableFuture 可以作為monad(單子)和functor。由于回調風格的實現，我們不必因為等待一個計算完成而阻塞著調用線程，而是告訴 CompletableFuture 當計算完成的時候請執行某個 function 。而且我們還可以將這些操作串聯起來，或者將 CompletableFuture 組合起來。 
public<U> CompletableFuture<U>thenApply(Function<?superT,? extends U> fn)
public<U> CompletableFuture<U>thenApplyAsync(Function<?superT,? extends U> fn)
public<U> CompletableFuture<U>thenApplyAsync(Function<?superT,? extends U> fn, Executor executor) 
  這一組函數的功能是當原來的CompletableFuture計算完后，將結果傳遞給函數 fn ，將 fn 的結果作為新的 CompletableFuture 計算結果。因此它的功能相當于將 CompletableFuture<T> 轉換成 CompletableFuture<U> 。 

 這三個函數的區別和上面介紹的一樣，不以 Async 結尾的方法由原來的線程計算，以 Async 結尾的方法由默認的線程池 ForkJoinPool.commonPool() 或者指定的線程池 executor 運行。Java的CompletableFuture類總是遵循這樣的原則，下面就不一一贅述了。 
使用例子如下：
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
return100;
});
CompletableFuture<String> f = future.thenApplyAsync(i -> i * 10).thenApply(i -> i.toString());
System.out.println(f.get()); //"1000" 
 需要注意的是，這些轉換并不是馬上執行的，也不會阻塞，而是在前一個stage完成后繼續執行。

 它們與 handle 方法的區別在于 handle 方法會處理正常計算值和異常，因此它可以屏蔽異常，避免異常繼續拋出。而 thenApply 方法只是用來處理正常值，因此一旦有異常就會拋出。 
純消費(執行Action)
 上面的方法是當計算完成的時候，會生成新的計算結果( thenApply , handle )，或者返回同樣的計算結果 whenComplete ， CompletableFuture 還提供了一種處理結果的方法，只對結果執行 Action ,而不返回新的計算值，因此計算值為 Void : 
publicCompletableFuture<Void>thenAccept(Consumer<?superT> action)
publicCompletableFuture<Void>thenAcceptAsync(Consumer<?superT> action)
publicCompletableFuture<Void>thenAcceptAsync(Consumer<?superT> action, Executor executor) 
  看它的參數類型也就明白了，它們是函數式接口 Consumer ，這個接口只有輸入，沒有返回值。 

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
return100;
});
CompletableFuture<Void> f = future.thenAccept(System.out::println);
System.out.println(f.get()); 
  
   thenAcceptBoth 以及相關方法提供了類似的功能，當兩個CompletionStage都正常完成計算的時候，就會執行提供的 action ，它用來組合另外一個異步的結果。 

 runAfterBoth 是當兩個CompletionStage都正常完成計算的時候,執行一個Runnable，這個Runnable并不使用計算的結果。 
</div>
public<U> CompletableFuture<Void>thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<?superT,?superU> action)
public<U> CompletableFuture<Void>thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<?superT,?superU> action)
public<U> CompletableFuture<Void>thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<?superT,?superU> action, Executor executor)
publicCompletableFuture<Void>runAfterBoth(CompletionStage<?> other, Runnable action) 
 例子如下：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
return100;
});
CompletableFuture<Void> f = future.thenAcceptBoth(CompletableFuture.completedFuture(10), (x, y) -> System.out.println(x * y));
System.out.println(f.get()); 
  更徹底地，下面一組方法當計算完成的時候會執行一個Runnable,與 thenAccept 不同，Runnable并不使用CompletableFuture計算的結果。 

publicCompletableFuture<Void>thenRun(Runnable action)
publicCompletableFuture<Void>thenRunAsync(Runnable action)
publicCompletableFuture<Void>thenRunAsync(Runnable action, Executor executor) 
  因此先前的CompletableFuture計算的結果被忽略了,這個方法返回 CompletableFuture<Void> 類型的對象。 

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
return100;
});
CompletableFuture<Void> f = future.thenRun(() -> System.out.println("finished"));
System.out.println(f.get()); 
  
   因此，你可以根據方法的參數的類型來加速你的記憶。 Runnable 類型的參數會忽略計算的結果， Consumer 是純消費計算結果， BiConsumer 會組合另外一個 CompletionStage 純消費， Function 會對計算結果做轉換， BiFunction 會組合另外一個 CompletionStage 的計算結果做轉換。 

</div>
組合
public<U> CompletableFuture<U>thenCompose(Function<?superT,? extends CompletionStage<U>> fn)
public<U> CompletableFuture<U>thenComposeAsync(Function<?superT,? extends CompletionStage<U>> fn)
public<U> CompletableFuture<U>thenComposeAsync(Function<?superT,? extends CompletionStage<U>> fn, Executor executor) 
 這一組方法接受一個Function作為參數，這個Function的輸入是當前的CompletableFuture的計算值，返回結果將是一個新的CompletableFuture，這個新的CompletableFuture會組合原來的CompletableFuture和函數返回的CompletableFuture。因此它的功能類似:

A +--> B +---> C 
  記住， thenCompose 返回的對象并不一是函數 fn 返回的對象，如果原來的 CompletableFuture 還沒有計算出來，它就會生成一個新的組合后的CompletableFuture。 

例子：
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
return100;
});
CompletableFuture<String> f = future.thenCompose( i -> {
returnCompletableFuture.supplyAsync(() -> {
return(i *10) +"";
 });
});
System.out.println(f.get()); //1000 
  而下面的一組方法 thenCombine 用來復合另外一個CompletionStage的結果。它的功能類似： 

 A +
  |
  +------> C
  +------^
B + 兩個CompletionStage是并行執行的，它們之間并沒有先后依賴順序， 
 other 并不會等待先前的 
 CompletableFuture 
 執行完畢后再執行。

public<U,V> CompletableFuture<V>thenCombine(CompletionStage<? extends U> other, BiFunction<?superT,?superU,? extends V> fn)
public<U,V> CompletableFuture<V>thenCombineAsync(CompletionStage<? extends U> other, BiFunction<?superT,?superU,? extends V> fn)
public<U,V> CompletableFuture<V>thenCombineAsync(CompletionStage<? extends U> other, BiFunction<?superT,?superU,? extends V> fn, Executor executor) 
  其實從功能上來講,它們的功能更類似 thenAcceptBoth ，只不過 thenAcceptBoth 是純消費，它的函數參數沒有返回值，而 thenCombine 的函數參數 fn 有返回值。 

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
return100;
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
return"abc";
});
CompletableFuture<String> f = future.thenCombine(future2, (x,y) -> y + "-"+ x);
System.out.println(f.get()); //abc-100 
 Either

 thenAcceptBoth 和 runAfterBoth 是當兩個CompletableFuture都計算完成，而我們下面要了解的方法是當任意一個CompletableFuture計算完成的時候就會執行。 
publicCompletableFuture<Void>acceptEither(CompletionStage<? extends T> other, Consumer<?superT> action)
publicCompletableFuture<Void>acceptEitherAsync(CompletionStage<? extends T> other, Consumer<?superT> action)
publicCompletableFuture<Void>acceptEitherAsync(CompletionStage<? extends T> other, Consumer<?superT> action, Executor executor)
public<U> CompletableFuture<U>applyToEither(CompletionStage<? extends T> other, Function<?superT,U> fn)
public<U> CompletableFuture<U>applyToEitherAsync(CompletionStage<? extends T> other, Function<?superT,U> fn)
public<U> CompletableFuture<U>applyToEitherAsync(CompletionStage<? extends T> other, Function<?superT,U> fn, Executor executor)</pre> 
 
 acceptEither 方法是當任意一個CompletionStage完成的時候， action 這個消費者就會被執行。這個方法返回 CompletableFuture<Void> 

 applyToEither 方法是當任意一個CompletionStage完成的時候， fn 會被執行，它的返回值會當作新的 CompletableFuture<U> 的計算結果。 

 下面這個例子有時會輸出 100 ,有時候會輸出 200 ,哪個Future先完成就會根據它的結果計算。 
Random rand = newRandom();
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
try{
 Thread.sleep(10000+ rand.nextInt(1000));
 } catch(InterruptedException e) {
 e.printStackTrace();
 }
return100;
});
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
try{
 Thread.sleep(10000+ rand.nextInt(1000));
 } catch(InterruptedException e) {
 e.printStackTrace();
 }
return200;
});
CompletableFuture<String> f = future.applyToEither(future2,i -> i.toString()); 
  輔助方法 allOf 和 anyOf 

 前面我們已經介紹了幾個靜態方法： completedFuture 、 runAsync 、 supplyAsync ,下面介紹的這兩個方法用來組合多個CompletableFuture。 
publicstaticCompletableFuture<Void>allOf(CompletableFuture<?>... cfs)
publicstaticCompletableFuture<Object>anyOf(CompletableFuture<?>... cfs) 
  
   allOf 方法是當所有的 CompletableFuture 都執行完后執行計算。 

 anyOf 方法是當任意一個 CompletableFuture 執行完后就會執行計算，計算的結果相同。 
</div>
 下面的代碼運行結果有時是100,有時是"abc"。但是 anyOf 和 applyToEither 不同。 anyOf 接受任意多的CompletableFuture但是 applyToEither 只是判斷兩個CompletableFuture, anyOf 返回值的計算結果是參數中其中一個CompletableFuture的計算結果， applyToEither 返回值的計算結果卻是要經過 fn 處理的。當然還有靜態方法的區別，線程池的選擇等。 
Random rand = newRandom();
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
try{
 Thread.sleep(10000+ rand.nextInt(1000));
 } catch(InterruptedException e) {
 e.printStackTrace();
 }
return100;
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
try{
 Thread.sleep(10000+ rand.nextInt(1000));
 } catch(InterruptedException e) {
 e.printStackTrace();
 }
return"abc";
});
//CompletableFuture<Void> f = CompletableFuture.allOf(future1,future2);
CompletableFuture<Object> f = CompletableFuture.anyOf(future1,future2);
System.out.println(f.get()); 
  我想通過上面的介紹，應該把CompletableFuture的方法和功能介紹完了( cancel 、 isCompletedExceptionally() 、 isDone() 以及繼承于Object的方法無需介紹了， toCompletableFuture() 返回CompletableFuture本身)，希望你能全面了解CompletableFuture強大的功能，并將它應用到Java的異步編程中。如果你有使用它的開源項目，可以留言分享一下。 

更進一步
 如果你用過Guava的Future類，你就會知道它的 Futures 輔助類提供了很多便利方法，用來處理多個Future，而不像Java的CompletableFuture，只提供了 allOf 、 anyOf 兩個方法。 比如有這樣一個需求，將多個CompletableFuture組合成一個CompletableFuture，這個組合后的CompletableFuture的計算結果是個List,它包含前面所有的CompletableFuture的計算結果，guava的 Futures.allAsList 可以實現這樣的功能，但是對于java CompletableFuture，我們需要一些輔助方法： 
publicstatic<T> CompletableFuture<List<T>>sequence(List<CompletableFuture<T>> futures) {
 CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futures.toArray(newCompletableFuture[futures.size()]));
returnallDoneFuture.thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.<T>toList()));
 }
publicstatic<T> CompletableFuture<Stream<T>>sequence(Stream<CompletableFuture<T>> futures) {
 List<CompletableFuture<T>> futureList = futures.filter(f -> f != null).collect(Collectors.toList());
returnsequence(futureList);
 } 
 或者Java Future轉CompletableFuture:

publicstatic<T> CompletableFuture<T>toCompletable(Future<T> future, Executor executor) {
returnCompletableFuture.supplyAsync(() -> {
try{
returnfuture.get();
 } catch(InterruptedException | ExecutionException e) {
thrownewRuntimeException(e);
 }
 }, executor);
} 
  github有多個項目可以實現Java CompletableFuture與其它Future (如Guava ListenableFuture)之間的轉換，如 spotify/futures-extra 、 future-converter 、 scala/scala-java8-compat 等。 

參考文檔
 
   Java 8: Definitive guide to CompletableFuture 

 https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html 
 https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html 
</ol>
 </div>
                    
                    
                         本文由用戶 yangtaochina 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。
                         轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。
                         本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！
                         本文地址：http://www.baiduhome.net/lib/view/open1456829642156.html
                         Java Java開發
                    

                
                
                    
                        相關經驗
                        
  Java CompletableFuture 詳解
   使用CompletableFuture處理異步超時
   java NIO詳解
   Java安全詳解
   Java Stream 詳解
                         
                    
                    
                        相關資訊
                        
  Java 7 Hotspot循環Bug詳解
   MySQL 5.7 新特性詳解
   詳解Android 4.2新特性
   NVIDIA Tegra 3 詳解
   Android 4.3 新特性詳解
                         
                    
                    
                        相關文檔
                        
   Java 同步詳解
    Java NIO 詳解
    java clone詳解
    Java數組Arrays詳解
    Java郵件開發詳解
    Java正則表達式入門+HTMLParser使用詳解
    SIP 協議詳解(中文)
    FreeMarker 深入淺出詳解
    Java最全筆試面試題(含答案)
                         
                    

                    目錄