Java之CompletableFuture异步、组合计算基本用法

悠悠 2023-06-11 07:27 10阅读 0赞

> CompletableFuture是Java8新增的Api,该类实现了Future和ComplateStage两个接口，提供了强大的Future扩展功能，可以简化异步编程的复杂性，提供了函数编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

*  Future:表示一个任务的生命周期，并提供了相应的方法来判断是否已经完成或取消，以及获取任务的结果和取消任务等；
 *  ComplateState:表示异步计算的一个阶段

##### CompletableFuture提供的静态方法 #####

*  返回运行在\{@link ForkJoinPool\#commonPool()\}中的异步计算任务，并返回一个包含Supplier计算结果的CompletableFuture新对象;

public static CompletableFuture supplyAsync(Supplier supplier)

示例：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{ 
 System.out.println("0----------");
 return 1;
 });
 System.out.println(future.get());

*  返回一个运行在给定的Executor中的异步任务，并返回一个包含调用Supplier返回结果的CompletableFuture:

public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)

示例：

ExecutorService executor = Executors.newFixedThreadPool(10);
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{ 
 System.out.println("0----------");
 return 1;
 }, executor);
 System.out.println(future.get());

*  返回一个运行在\{@link ForkJoinPool\#commonPool()\}中的异步任务，并且返回一个新的CompletableFuture:

public static CompletableFuture<Void> runAsync(Runnable runnable)

示例：

CompletableFuture<Void> future = CompletableFuture.runAsync(()->{ 
 System.out.println("--------------test-------------");
 });
 //返回值为null
 System.out.println(future.get());

*  返回一个运行在给定的Executor中的异步任务，并且返回一个新的CompletableFuture:

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

示例：

ExecutorService executor = Executors.newFixedThreadPool(10);
 CompletableFuture<Void> future = CompletableFuture.runAsync(()->{ 
 System.out.println("--------------test-------------");
 }, executor);
 //返回值null
 System.out.println(future.get());

*  创建一个有初始值的CompletableFuture

public static CompletableFuture completedFuture(U value)

示例：

CompletableFuture<Integer> future = CompletableFuture.completedFuture(1);
 System.out.println(future.get());

*  当给定的所有CompletableFuture执行完成返回一个新的CompletableFuture。如果给定的任何一个CompletionFuture执行异常，那么返回的CompletionFuture也异常，因此使用CompletionException异常块包括这段代码,返回的值不包括任何执行的结果。

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

示例：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return 1;
 });
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 String s = null;
 s.length();
 return 2;
 });
 CompletableFuture<Void> future2 = CompletableFuture.allOf(future, future1);
 System.out.println(future2.get());

*  当给定的任何一个CompletableFuture执行完成就返回执行结果，如果发生异常则抛出异常

CompletableFuture<Object> future2 = CompletableFuture.anyOf(
 CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 try { 
 TimeUnit.MILLISECONDS.sleep(10);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 }
 String s = null;
 s.length();
 return 1;
 }),
 CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 }),
 CompletableFuture.supplyAsync(()->{ 
 System.out.println("------3------------");
 return 3;
 })
 );
 System.out.println(future2.get());

### 实例方法 ###

*  thenRun/thenRunAsync

上面的任务执行完成后会同步执行，会阻塞到执行完毕，不会将上一步的结果传递过来

public CompletableFuture<Void> thenRun(Runnable action)

CompletableFuture<Void> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 }).thenRun(()->{ 
 System.out.println("------runnable----------");
 });

上一个任务执行完成后会异步执行，不会阻塞，也不会将上一个任务的执行结果传递过来

public CompletableFuture<Void> thenRunAsync(Runnable action)

public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

示例：

CompletableFuture<Void> future3 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------3------------");
 return 3;
 }).thenRunAsync(()->{ 
 System.out.println("---------runAsync-----------");
 });

*  thenAccept /thenAcceptAsync

/** * 上一个任务执行完成后，将执行结果作为参数传递过来，同步执行 **/
 public CompletableFuture<Void> thenAccept(Consumer<? super T> action)

示例：

CompletableFuture<Void> future = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------3------------");
 return 3;
 }).thenAccept((age)->{ 
 System.out.println("接收到的参数是："+age);
 });
 future.get();

/** * 上一个任务执行完成后，将结果作为参数传递过来，异步执行 **/
 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
 Executor executor)

示例：

CompletableFuture<Void> future = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------3------------");
 return 3;
 }).thenAcceptAsync((age)->{ 
 System.out.println("接收到的参数是："+age);
 });
 future.get();

*  thenAcceptBoth

/** * 利用前两个任务的计算结果作为参数，做同步计算 **/
 public CompletableFuture<Void> thenAcceptBoth(
 CompletionStage<? extends U> other,
 BiConsumer<? super T, ? super U> action)

示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return "1";
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<Void> future3 = future1.thenAcceptBoth(future2, (a, b)->{ 
 System.out.println("执行结果是：("+a+b+")");
 });
 future3.get();

/** * 利用前两个任务的计算结果作为参数，异步计算 **/
 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)

示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return "1";
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<Void> future3 = future1.thenAcceptBothAsync(future2, (a, b)->{ 
 System.out.println("执行结果是：("+a+b+")");
 });
 future3.get();

*  thenCombine/thenCombineAsync

/** * 利用前两个任务的结果作为参数做计算，同步，有返回值 **/
 public <U,V> CompletableFuture<V> thenCombine(
 CompletionStage<? extends U> other,
 BiFunction<? super T,? super U,? extends V> fn)

示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return "1";
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<String> future3 = future1.thenCombineAsync(future2, (a, b)->{ 
 return "a+b="+a+b;
 });
 future3.get();

/** *利用前两个任务的计算结果作为参数，异步计算，返回结果 **/
 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)

示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return "1";
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<String> future3 = future1.thenCombineAsync(future2, (a, b)->{ 
 return "a+b="+a+b;
 });
 future3.get();

*  thenCompose/thenComposeAsync

/** * 该方法允许你对两个任务进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个任务 **/
 public CompletableFuture thenCompose(
 Function<? super T, ? extends CompletionStage> fn)

示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return "1";
 });
 CompletableFuture<String> future3 = future1.thenCompose((a)->{ 
 return CompletableFuture.completedFuture(a);
 });

/** * 该方法允许你对两个任务进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个任务，异步执* 行，非阻塞； **/
 public CompletableFuture thenComposeAsync(
 Function<? super T, ? extends CompletionStage> fn)
 public CompletableFuture thenComposeAsync(
 Function<? super T, ? extends CompletionStage> fn,
 Executor executor)

示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return "1";
 });
 CompletableFuture<String> future3 = future1.thenComposeAsync((a)->{ 
 return CompletableFuture.completedFuture(a);
 });

*  thenApply/thenApplyAsync

/** * 当一个任务依赖于另外一个任务的返回结果是，可以将返回结果作为参数传递，并返回执行后的结果 **/
 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)

*  handle/handleAsync

/** * 任务执行完成后对结果进行处理，支持异常处理 **/
 public CompletableFuture handle(
 BiFunction<? super T, Throwable, ? extends U> fn)
 
 public CompletableFuture handleAsync(
 BiFunction<? super T, Throwable, ? extends U> fn)
 
 public CompletableFuture handleAsync(
 BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

示例：

*  applyToEither

/** * 两个任务谁执行的快，直径的结果作为参数参与下一步的计算 **/
 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);

示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return 1;
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<String> future3 = future1.applyToEither(future2, (a)->{ 
 return "a"+a;
 });

*  acceptEither

/** * 两个任务谁执行的快，就使用返回的结果进行下一不计算，无计算返回值 **/
 public CompletableFuture<Void> acceptEither(
 CompletionStage<? extends T> other, Consumer<? super T> action)
 public CompletableFuture<Void> acceptEitherAsync(
 CompletionStage<? extends T> other, Consumer<? super T> action)
 public CompletableFuture<Void> acceptEitherAsync(
 CompletionStage<? extends T> other, Consumer<? super T> action,
 Executor executor)

示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------1------------");
 return 1;
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<Void> future3 = future1.acceptEither(future2, (a)->{ 
 System.out.println("-----future3----------");
 });

*  isCancelled

/** * 如果CompletableFuture在正常完成之前被取消，则返回true **/
    public boolean isCancelled()

*  isVompletedExceptionally

/** * 如果CompletableFuture执行异常，无论什么原因，则返回true **/
    public boolean isCompletedExceptionally()

*  isDone

/** * 如果CompletableFutrue执行完成，正常、异常、关闭，都返回true **/
    public boolean isDone()

*  cancel

/** * 如果CompletableFuture还未完成，则会关闭 **/
    public boolean cancel(boolean mayInterruptIfRunning)

*  complete

/** * 立即完成计算，设定指定的结果，并返回计算结果 **/
    public boolean complete(T value)

示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{ 
 try { 
 TimeUnit.SECONDS.sleep(12);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 }
 System.out.println("------1------------");
 return 1;
 });
 boolean future3 = future1.complete(4);
 System.out.println("--------------");
 System.out.println(future1.get());
 System.out.println("--------------");

*  completeExceptionally

/** * 立即完成计算，并抛出异常 **/
    public boolean completeExceptionally(Throwable ex)

*  runAfterBoth

/** * 两个任务执行完成之后，才会执行接下来的任务 **/
 public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor);

示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{ 
 try { 
 TimeUnit.SECONDS.sleep(12);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 }
 System.out.println("------1------------");
 return 1;
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<Void> future3 = future1.runAfterBoth(future2, ()->{ 
 System.out.println("--------------3-----------"); 
 });

*  runAfterEither

/** * 两个任务，任何一个执行完成都会执行下一步操作 **/
 public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor executor);

示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{ 
 try { 
 TimeUnit.SECONDS.sleep(12);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 }
 System.out.println("------1------------");
 return 1;
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{ 
 System.out.println("------2------------");
 return 2;
 });
 CompletableFuture<Void> future3 = future1.runAfterEither(future2, ()->{ 
 System.out.println("--------------3-----------");
 });

*  whenComplete

/** * 当任务计算完成，或者抛出异常的时候，可以执行特定的任务 **/
 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)

示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->			{ 
 System.out.println("------1------------");
 return 1;
 });
 future1.whenComplete((t, throwable)->{ 
 System.out.println("--------------3--------");
 });