异步方法CompletableFuture的使用

- 日理万妓 2022-10-22 02:57 155阅读 0赞

### 文章目录 ###

*  一、创建异步对象
 *  二、计算完成时回调方法
 *  三、handle方法
 *  四、线程串行化方法
 *  五、两任务组合
 *  六、多任务组合

`Future`是 `JDK5`添加的类，用来描述一个异步计算的结果。你可以使用 `isDone`方法检查计算是否完成，或者使用 `get`阻塞住调用线程，直到计算完成返回结果，你也可以使用 `cancel`方法停止任务的执行。虽然 `Future`以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的 `CPU`资源，而且也不能及时地得到计算结果，为什么不能用观察者设计模式当计算结果完成及时通知监听者呢？

`JDK8`新增加了一个`CompletableFuture`类，提供了非常强大的`Future`的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合`CompletableFuture`的方法。`CompletableFuture`类实现了`Future`接口，所以你还是可以像以前一样通过`get`方法阻塞或者轮询的方式获得结果，但是这种方式不推荐使用。`CompletableFuture`和`FutureTask`同属于`Future`接口的实现类，都可以获取线程的执行结果。

# 一、创建异步对象 #

`CompletableFuture`提供了四个静态方法来创建一个异步操作：

static CompletableFuture<Void> runAsync(Runnable runnable)
 public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
 public static CompletableFuture supplyAsync(Supplier supplier)
 public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)

没有指定`Executor`的方法会使用`ForkJoinPool.commonPool()`作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。以下所有的方法都类同。

*  `runAsync`方法不支持返回值；
 *  `supplyAsync`可以支持返回值。

# 二、计算完成时回调方法 #

当`CompletableFuture`的计算结果完成，或者抛出异常的时候，可以执行特定的`Action`。主要是下面的方法：

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

`whenComplete`可以处理正常和异常的计算结果，`exceptionally`处理异常情况。`BiConsumer<? super T,? super Throwable>`可以定义处理业务。

`whenComplete`是同执行当前任务的线程执行继续执行`whenComplete`的任务；`whenCompleteAsync`是执行把`whenCompleteAsync`这个任务继续提交给线程池来进行执行。

示例代码如下：

public static void main(String[] args) throws Exception { 
 System.out.println("---异步方法执行前---");
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { 
 System.out.println("---异步方法执行中---");
 return 1;
 }).whenComplete((res, err) -> { 
 System.out.println("---异步任务执行完成---，返回结果" + res);
 });
 System.out.println("---获取返回结果为--- " + future.get());
 System.out.println("---异步方法执行后---");
 }

# 三、handle方法 #

`handle`是执行任务完成时对结果的处理。`handle`是在任务完成后再执行，同时可以处理异常的任务。

示例代码如下：

public static void main(String[] args) throws Exception { 
 System.out.println("---异步方法执行前---");
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { 
 System.out.println("---异步方法执行中---");
 return 1;
 }).handle((res, err) -> { 
 System.out.println("---异步任务执行完成后执行handle方法---");
 return 2;
 }).whenComplete((res, err) -> { 
 System.out.println("---异步任务执行完成---，返回结果" + res);
 });
 System.out.println("---获取返回结果为--- " + future.get());
 System.out.println("---异步方法执行后---");
 }

# 四、线程串行化方法 #

*  `thenApply()`：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值。
 *  `thenAccept()`：消费处理结果。接收任务的处理结果，并消费处理，无返回结果。
 *  `thenRun()`：只要上面的任务执行完成，就开始执行`thenRun`，只是处理完任务后，执行`henRun`的后续操作

带有`Async`默认是异步执行的，这里所谓的异步指的是不在当前线程内执行。

public CompletableFuture thenApply(Function<? super T,? extends U> fn)
 public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn)
 public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
 
 public CompletionStage<Void> thenAccept(Consumer<? super T> action);
 public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
 public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);
 
 public CompletionStage<Void> thenRun(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

> 上面方法中 T：上一个任务返回结果的类型，U：当前任务的返回值类型

示例代码如下：

# 五、两任务组合 #

两个任务必须都完成，触发该任务。

*  `thenCombine`：组合两个`future`，获取两个`future`的返回结果，并返回当前任务的返回值；
 *  `thenAcceptBoth`：组合两个`future`，获取两个`future`任务的返回结果，然后处理任务，没有返回值；
 *  `runAfterBoth`：组合两个`future`，不需要获取`future`的结果，只需两个`future`处理完任务后，处理该任务。

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);
 
 
 public CompletableFuture<Void> thenAcceptBoth(
 CompletionStage<? extends U> other,
 BiConsumer<? super T, ? super U> action);
 
 public CompletableFuture<Void> thenAcceptBothAsync(
 CompletionStage<? extends U> other,
 BiConsumer<? super T, ? super U> action);
 
 public CompletableFuture<Void> thenAcceptBothAsync(
 CompletionStage<? extends U> other,
 BiConsumer<? super T, ? super U> action, Executor executor);
 
 
 public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
 Runnable action);
 
 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
 Runnable action);
 
 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
 Runnable action,
 Executor executor);

示例代码如下：

public static void main(String[] args) throws Exception { 
            CompletableFuture.supplyAsync(() -> { 
                try { 
                    TimeUnit.SECONDS.sleep(2);
                    System.out.println("返回a");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                return "a";
            }).applyToEitherAsync(CompletableFuture.supplyAsync(() -> { 
                try { 
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println("返回b");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                return "b";
            }), res -> res).whenCompleteAsync((res, err) -> System.out.println("最快返回：" + res));
            TimeUnit.SECONDS.sleep(10);
        }

打印：

返回a
    最快返回：a
    返回b

# 六、多任务组合 #

// 等待所有任务完成
 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs);
 // 只要有一个任务完成
 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs);

示例代码：

public static void main(String[] args) throws Exception { 
 List<CompletableFuture<String>> futures = Arrays.asList(CompletableFuture.completedFuture("hello"),
 CompletableFuture.completedFuture("world"),CompletableFuture.completedFuture("me"));
 CompletableFuture<Void> allCompleted = CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{ }));
 allCompleted.thenRun(() -> { 
 futures.stream().forEach(future -> { 
 try { 
 System.out.println("get future at:" + System.currentTimeMillis() + ", result:" + future.get());
 } catch (Exception e) { 
 e.printStackTrace();
 }
 });
 });
 }