[大数据]CompletableFuture详解~创建实例

创建 CompletableFuture 对象实例我们可以使用如下几个方法：

static CompletableFuture<U> completedFuture(U value)

//使用forkjoin公共线程池
static CompletableFuture<Void> runAsync(Runnable runnable)
static CompletableFuture<U> supplyAsync(Supplier<U> supplier)

//使用自定义线程池
static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
static CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

第一个方法创建一个具有默认结果的 CompletableFuture，这个没啥好讲。我们重点讲述下下面四个异步方法。

前两个方法 runAsync 不支持返回值，而 supplyAsync可以支持返回结果。

这个两个方法默认将会使用公共的?ForkJoinPool?线程池执行，这个线程池默认线程数是?CPU?的核数。

使用共享线程池将会有个弊端，一旦有任务被阻塞，将会造成其他任务没机会执行。所以强烈建议使用后两个方法，根据任务类型不同，主动创建线程池，进行资源隔离，避免互相干扰。

-------------------------------------------------------

CompletableFuture的创建：

• 说明：

  • 两个重载方法之间的区别 => 后者可以传入自定义Executor，前者是默认的，使用的ForkJoinPool

  • supplyAsync和runAsync方法之间的区别 => 前者有返回值，后者无返回值

  • Supplier是函数式接口，因此该方法需要传入该接口的实现类，追踪源码会发现在run方法中会调用该接口的方法。因此使用该方法创建CompletableFuture对象只需重写Supplier中的get方法，在get方法中定义任务即可。又因为函数式接口可以使用Lambda表达式，和new创建CompletableFuture对象相比代码会简洁不少

  • 使用new方法

CompletableFuture<Double> futurePrice = new CompletableFuture<>();

使用CompletableFuture#completedFuture静态方法创建

public static <U> CompletableFuture<U> completedFuture(U value) {
    return new CompletableFuture<U>((value == null) ? NIL : value);
}

• 参数的值为任务执行完的结果，一般该方法在实际应用中较少应用

• 使用 CompletableFuture#supplyAsync静态方法创建 supplyAsync有两个重载方法：

//方法一
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}
//方法二
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
                                                   Executor executor) {
    return asyncSupplyStage(screenExecutor(executor), supplier);
}

• 使用CompletableFuture#runAsync静态方法创建 runAsync有两个重载方法

//方法一
public static CompletableFuture<Void> runAsync(Runnable runnable) {
    return asyncRunStage(asyncPool, runnable);
}
//方法二
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) {
    return asyncRunStage(screenExecutor(executor), runnable);
}

结果的获取：?对于结果的获取CompltableFuture类提供了四种方式

//方式一
public T get()
//方式二
public T get(long timeout, TimeUnit unit)
//方式三
public T getNow(T valueIfAbsent)
//方式四
public T join()

• 说明：

  示例：

  • get()和get(long timeout, TimeUnit unit) => 在Future中就已经提供了，后者提供超时处理，如果在指定时间内未获取结果将抛出超时异常

  • getNow => 立即获取结果不阻塞，结果计算已完成将返回结果或计算过程中的异常，如果未计算完成将返回设定的valueIfAbsent值

  • join => 方法里不会抛出异常

public class AcquireResultTest {
  public static void main(String[] args) throws ExecutionException, InterruptedException {
      //getNow方法测试
      CompletableFuture<String> cp1 = CompletableFuture.supplyAsync(() -> {
          try {
              Thread.sleep(60 * 1000 * 60 );
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
  
          return "hello world";
      });
  
      System.out.println(cp1.getNow("hello h2t"));
  
      //join方法测试
      CompletableFuture<Integer> cp2 = CompletableFuture.supplyAsync((()-> 1 / 0));
      System.out.println(cp2.join());
  
      //get方法测试
      CompletableFuture<Integer> cp3 = CompletableFuture.supplyAsync((()-> 1 / 0));
      System.out.println(cp3.get());
  }
}

说明：

• 第一个执行结果为hello h2t，因为要先睡上1分钟结果不能立即获取

• join方法获取结果方法里不会抛异常，但是执行结果会抛异常，抛出的异常为CompletionException

• get方法获取结果方法里将抛出异常，执行结果抛出的异常为ExecutionException

• 异常处理：?使用静态方法创建的CompletableFuture对象无需显示处理异常，使用new创建的对象需要调用completeExceptionally方法设置捕获到的异常，举例说明：

CompletableFuture completableFuture = new CompletableFuture();
new Thread(() -> {
   try {
       //doSomething，调用complete方法将其他方法的执行结果记录在completableFuture对象中
       completableFuture.complete(null);
   } catch (Exception e) {
       //异常处理
       completableFuture.completeExceptionally(e);
    }
}).start();