Java8新的異步編程方式CompletableFuture實(shí)現(xiàn)

一. Future

JDK 5引入了Future模式。Future接口是Java多線程Future模式的實(shí)現(xiàn)，在java.util.concurrent包中，可以來(lái)進(jìn)行異步計(jì)算。

Future模式是多線程設(shè)計(jì)常用的一種設(shè)計(jì)模式。Future模式可以理解成：我有一個(gè)任務(wù)，提交給了Future，F(xiàn)uture替我完成這個(gè)任務(wù)。期間我自己可以去做任何想做的事情。一段時(shí)間之后，我就便可以從Future那兒取出結(jié)果。

Future的接口很簡(jiǎn)單，只有五個(gè)方法。

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) 取消任務(wù)的執(zhí)行。參數(shù)指定是否立即中斷任務(wù)執(zhí)行，或者等等任務(wù)結(jié)束
• boolean isCancelled () 任務(wù)是否已經(jīng)取消，任務(wù)正常完成前將其取消，則返回 true
• boolean isDone () 任務(wù)是否已經(jīng)完成。需要注意的是如果任務(wù)正常終止、異?；蛉∠?，都將返回true
• V get () throws InterruptedException, ExecutionException 等待任務(wù)執(zhí)行結(jié)束，然后獲得V類型的結(jié)果。InterruptedException 線程被中斷異常， ExecutionException任務(wù)執(zhí)行異常，如果任務(wù)被取消，還會(huì)拋出CancellationException
• V get (long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException 同上面的get功能一樣，多了設(shè)置超時(shí)時(shí)間。參數(shù)timeout指定超時(shí)時(shí)間，uint指定時(shí)間的單位，在枚舉類TimeUnit中有相關(guān)的定義。如果計(jì) 算超時(shí)，將拋出TimeoutException

一般情況下，我們會(huì)結(jié)合Callable和Future一起使用，通過(guò)ExecutorService的submit方法執(zhí)行Callable，并返回Future。

  ExecutorService executor = Executors.newCachedThreadPool();
 
  Future<String> future = executor.submit(() -> { //Lambda 是一個(gè) callable， 提交后便立即執(zhí)行，這里返回的是 FutureTask 實(shí)例
   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 在其他線程中運(yùn)行著，可以做一些其他的事情
 
  try {
   System.out.println(future.get()); //等待 future 的執(zhí)行結(jié)果，執(zhí)行完畢之后打印出來(lái)
  } catch (InterruptedException e) {
  } catch (ExecutionException e) {
 
  } finally {
   executor.shutdown();
  }

比起future.get()，其實(shí)更推薦使用get (long timeout, TimeUnit unit) 方法，設(shè)置了超時(shí)時(shí)間可以防止程序無(wú)限制的等待future的結(jié)果。

二. CompletableFuture介紹

2.1 Future模式的缺點(diǎn)

Future雖然可以實(shí)現(xiàn)獲取異步執(zhí)行結(jié)果的需求，但是它沒(méi)有提供通知的機(jī)制，我們無(wú)法得知Future什么時(shí)候完成。

要么使用阻塞，在future.get()的地方等待future返回的結(jié)果，這時(shí)又變成同步操作。要么使用isDone()輪詢地判斷Future是否完成，這樣會(huì)耗費(fèi)CPU的資源。

2.2 CompletableFuture介紹

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

Java 8新增的CompletableFuture類正是吸收了所有Google Guava中ListenableFuture和SettableFuture的特征，還提供了其它強(qiáng)大的功能，讓Java擁有了完整的非阻塞編程模型：Future、Promise 和 Callback(在Java8之前，只有無(wú)Callback 的Future)。

CompletableFuture能夠?qū)⒒卣{(diào)放到與任務(wù)不同的線程中執(zhí)行，也能將回調(diào)作為繼續(xù)執(zhí)行的同步函數(shù)，在與任務(wù)相同的線程中執(zhí)行。它避免了傳統(tǒng)回調(diào)最大的問(wèn)題，那就是能夠?qū)⒖刂屏鞣蛛x到不同的事件處理器中。

CompletableFuture彌補(bǔ)了Future模式的缺點(diǎn)。在異步的任務(wù)完成后，需要用其結(jié)果繼續(xù)操作時(shí)，無(wú)需等待?？梢灾苯油ㄟ^(guò)thenAccept、thenApply、thenCompose等方式將前面異步處理的結(jié)果交給另外一個(gè)異步事件處理線程來(lái)處理。

三. CompletableFuture特性

3.1 CompletableFuture的靜態(tài)工廠方法

runAsync 和 supplyAsync 方法的區(qū)別是runAsync返回的CompletableFuture是沒(méi)有返回值的。    

 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

future.get()在等待執(zhí)行結(jié)果時(shí)，程序會(huì)一直block，如果此時(shí)調(diào)用complete(T t)會(huì)立即執(zhí)行。

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

執(zhí)行結(jié)果：

World

可以看到future調(diào)用complete(T t)會(huì)立即執(zhí)行。但是complete(T t)只能調(diào)用一次，后續(xù)的重復(fù)調(diào)用會(huì)失效。

如果future已經(jīng)執(zhí)行完畢能夠返回結(jié)果，此時(shí)再調(diào)用complete(T t)則會(huì)無(wú)效。

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

執(zhí)行結(jié)果：

Hello

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

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

執(zhí)行結(jié)果：

java.util.concurrent.ExecutionException: java.lang.Exception
...

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持億速云。