Java8 CompletableFuture 编程

朱雀 2020-11-07 01:57 625阅读 0赞

## 一、简介 ##

所谓异步调用其实就是实现一个无需等待被调用函数的返回值而让操作继续运行的方法。在 Java 语言中，简单的讲就是另启一个线程来完成调用中的部分计算，使调用继续运行或返回，而不需要等待计算结果。但调用者仍需要取线程的计算结果。

JDK5新增了 Future 接口，用于描述一个异步计算的结果。虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的 CPU 资源，而且也不能及时地得到计算结果。

private static final ExecutorService POOL = Executors.newFixedThreadPool(TASK_THRESHOLD, new ThreadFactory() { AtomicInteger atomicInteger = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { return new Thread(r, "demo15-" + atomicInteger.incrementAndGet()); } }); public static void main(String[] args) throws ExecutionException, InterruptedException { Future<Integer> submit = POOL.submit(() -> 123); // 1. get() 方法用户返回计算结果，如果计算还没有完成，则在get的时候会进行阻塞，直到获取到结果为止 Integer get = submit.get(); // 2. isDone() 方法用于判断当前Future是否执行完成。 boolean done = submit.isDone(); // 3. cancel(boolean mayInterruptIfRunning) 取消当前线程的执行。参数表示是否在线程执行的过程中阻断。 boolean cancel = submit.cancel(true); // 4. isCancelled() 判断当前task是否被取消. boolean cancelled = submit.isCancelled(); // 5. invokeAll 批量执行任务 Callable<String> callable = () -> "Hello Future"; List<Callable<String>> callables = Lists.newArrayList(callable, callable, callable, callable); List<Future<String>> futures = POOL.invokeAll(callables); }

在Java8中，CompletableFuture 提供了非常强大的 Future 的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合 CompletableFuture 的方法。

**tips：** CompletionStage 代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段。

## 二、CompletableFuture 使用 ##

### 1. runAsync、supplyAsync ###

// 无返回值 public 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)

runAsync、supplyAsync 方法是 CompletableFuture 提供的创建异步操作的方法。需要注意的是，如果没有指定 Executor 作为线程池，将会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码；如果指定线程池，则使用指定的线程池运行。以下所有的方法都类同。

public class Demo1 { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> System.out.println(123)); CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> "CompletableFuture"); System.out.println(supplyAsync.get()); } }

### 2. whenComplete、exceptionally ###

// 执行完成时，当前任务的线程执行继续执行 whenComplete 的任务。 public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action) // 执行完成时，把 whenCompleteAsync 这个任务提交给线程池来进行执行。 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)

当 CompletableFuture 的计算完成时，会执行 whenComplete 方法；当 CompletableFuture 计算中抛出异常时，会执行 exceptionally 方法。

public class Demo2 { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<Integer> runAsync = CompletableFuture.supplyAsync(() -> 123456); runAsync.whenComplete((t, throwable) -> { System.out.println(t); if (throwable != null) { throwable.printStackTrace(); } }); runAsync.whenCompleteAsync((t, throwable) -> { System.out.println(t); if (throwable != null) { throwable.printStackTrace(); } }); runAsync.exceptionally((throwable) -> { if (throwable != null) { throwable.printStackTrace(); } return null; }); TimeUnit.SECONDS.sleep(2); } }

### 3. thenApply、handle ###

// T：上一个任务返回结果的类型 // U：当前任务的返回值类型 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 handle(BiFunction<? super T, Throwable, ? extends U> fn); public CompletionStage handleAsync(BiFunction<? super T, Throwable, ? extends U> fn); public CompletionStage handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);

当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化

handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行，还可以处理异常的任务。thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法。

public class Demo3 { public static void main(String[] args) throws ExecutionException, InterruptedException { // thenApply CompletableFuture<Integer> thenApply = CompletableFuture.supplyAsync(() -> 123).thenApply(t -> t * t); System.out.println(thenApply.get()); // handle CompletableFuture<Integer> handle = CompletableFuture.supplyAsync(() -> { int i = 10 / 0; return new Random().nextInt(10); }).handle((t, throwable) -> { if (throwable != null) { throwable.printStackTrace(); return -1; } return t * t; }); System.out.println(handle.get()); } }

### 4. thenAccept、thenRun ###

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

thenAccept 接收任务的处理结果，并消费处理。无返回结果。

thenRun 跟 thenAccept 方法不一样的是，不关心任务的处理结果。只要上面的任务执行完成，就开始执行 thenRun。

public class Demo4 { public static void main(String[] args) { // thenAccept CompletableFuture<Void> thenAccept = CompletableFuture.supplyAsync(() -> new Random().nextInt(10)).thenAccept(System.out::println); // thenRun CompletableFuture<Void> thenRun = CompletableFuture.supplyAsync(() -> new Random().nextInt(10)).thenRun(() -> System.out.println(123)); } }

### 5. thenCombine、thenAcceptBoth ###

// T 表示第一个 CompletionStage 的返回结果类型 // U 表示第二个 CompletionStage 的返回结果类型 // V表示 thenCombine/thenAcceptBoth 处理结果类型 public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn); public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn); public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn,Executor executor); public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn); public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn); public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn,Executor executor);

thenCombine、thenAcceptBoth 都是用来合并任务 —— 等待两个 CompletionStage 的任务都执行完成后，把两个任务的结果一并来处理。区别在于 thenCombine 有返回值；thenAcceptBoth 无返回值。

public class Demo5 { public static void main(String[] args) throws ExecutionException, InterruptedException { // thenCombine CompletableFuture<String> thenCombine = CompletableFuture.supplyAsync(() -> new Random().nextInt(10)) .thenCombine(CompletableFuture.supplyAsync(() -> "str"), // 第一个参数是第一个 CompletionStage 的处理结果 // 第二个参数是第二个 CompletionStage 的处理结果 (i, s) -> i + s ); System.out.println(thenCombine.get()); // thenAcceptBoth CompletableFuture<Void> thenAcceptBoth = CompletableFuture.supplyAsync(() -> new Random().nextInt(10)) .thenAcceptBoth(CompletableFuture.supplyAsync(() -> "str"), (i, s) -> System.out.println(i + s)); } }

### 6. applyToEither、acceptEither、runAfterEither、runAfterBoth ###

*  applyToEither：两个 CompletionStage，谁执行返回的结果快，就用那个 CompletionStage 的结果进行下一步的处理，有返回值。
 *  acceptEither：两个 CompletionStage，谁执行返回的结果快，就用那个 CompletionStage 的结果进行下一步的处理，无返回值。
 *  runAfterEither：两个 CompletionStage，任何一个完成了，都会执行下一步的操作（Runnable），无返回值。
 *  runAfterBoth：两个 CompletionStage，都完成了计算才会执行下一步的操作（Runnable），无返回值。

由于这几个方法含义相近，使用更加类似，我们就以 applyToEither 来介绍...

// T 两个 CompletionStage 组合运算后的结果类型 // U 下一步处理运算的结果返回值类型 public CompletionStage applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn); public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn); public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn,Executor executor);

public class Demo6 { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<Integer> applyToEither = CompletableFuture.supplyAsync(() -> { int nextInt = new Random().nextInt(10); try { Thread.sleep(nextInt); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("f1=" + nextInt); return nextInt; }).applyToEither(CompletableFuture.supplyAsync(() -> { int nextInt = new Random().nextInt(10); try { Thread.sleep(nextInt); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("f2=" + nextInt); return nextInt; }), i -> i); System.out.println(applyToEither.get()); } }

### 7. thenCompose ###

public CompletableFuture thenCompose(Function<? super T, ? extends CompletionStage> fn); public CompletableFuture thenComposeAsync(Function<? super T, ? extends CompletionStage> fn) ; public CompletableFuture thenComposeAsync(Function<? super T, ? extends CompletionStage> fn, Executor executor) ;

thenCompose 方法允许你对两个 CompletionStage 进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个操作。

public class Demo7 { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<Integer> thenCompose = CompletableFuture.supplyAsync(() -> new Random().nextInt(10)) .thenCompose(i -> CompletableFuture.supplyAsync(() -> i * i)); System.out.println(thenCompose.get()); } }