CompleteFuture 学习

一 CompleteFuture 使用场景

• 创建异步任务
• 简单任务异步回调
• 多任务组合处理

二 创建异步任务

• supplyAsync (有返回值)

//使用默认线程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) 
//使用自定义线程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) 

• runAsync（无返回值）

//使用默认线程池
public static CompletableFuture<Void> runAsync(Runnable runnable)
//使用自定义线程池
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

三 任务异步回调

1. thenRun/thenRunAsync

上一个任务完成后执行

//不关心上一个任务执行返回结果
//thenRun 使用上一个任务的线程池
public CompletableFuture<Void> thenRun(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor)

2. thenAccept/thenAcceptAsync

第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，但是回调方法是没有返回值的。

public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)

3. thenApply/thenApplyAsync

第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，并且回调方法是有返回值的。

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

4. exceptionally

某个任务执行异常时，执行的回调方法;并且有抛出异常作为参数，传递到回调方法

 public CompletableFuture<T> exceptionally(
        Function<Throwable, ? extends T> fn) {
        return uniExceptionallyStage(fn);
    }

5. whenComplete

某个任务执行完成后，执行的回调方法；并且whenComplete方法返回的CompletableFuture的result是上个任务的结果。

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)

6. handle

某个任务执行完成后，执行回调方法，并且是有返回值的;并且handle方法返回的CompletableFuture的result是回调方法执行的结果。

public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) 
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

测试Async

 public void testAsync() throws ExecutionException, InterruptedException, TimeoutException {
        //thenRun
        CompletableFuture.runAsync(()->{
            log.info(String.valueOf(System.currentTimeMillis()));
            //模拟任务执行，耗时10s
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            log.info("---thenRun---+++++++++++ 执行Task1 +++++++++++"+Thread.currentThread().getName());
                },executorConfig.ThreadPoolTaskExecutor()
        ).thenRun(()->{
            log.info("---thenRun---+++++++++++ Task1执行完成后执行Task2 +++++++++++"+Thread.currentThread().getName());
        });
        //thenAccept
        CompletableFuture.supplyAsync(()->{
            //模拟任务执行，耗时10s
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            log.info("---thenAccept---+++++++++++ 执行Task1并返回 +++++++++++"+Thread.currentThread().getName());
           return  "result of Task1";
        }).thenAccept((result)->{
            //Task1完成后，Task2接收到Task1的结果进行处理
            String resultUpperCase=result.toUpperCase();
            log.info("---thenAccept---Task2接收到Task1的结果并处理： "+resultUpperCase+Thread.currentThread().getName());
        });

        //thenApply
        CompletableFuture<Integer> apply = CompletableFuture.supplyAsync(() -> {
            log.info("---thenApply---+++++++++++ 执行Task1 +++++++++++"+Thread.currentThread().getName());
            //Task1执行完成后返回结果
            return 1;
        }, executorConfig.ThreadPoolTaskExecutor()).thenApply((r) -> {
            //接收Task1结果，处理并返回
            log.info("---thenApply--- Taks2收到Taks1结果处理后返回： "+ r);
            return r + 1;
        });
        //获取Task2最终返回结果
        int task2Result= apply.get(10, TimeUnit.SECONDS);
        System.out.println("获取到Task2的结果： "+task2Result);


    }

四 多任务组合处理

1 AND

将两个CompletableFuture组合起来，只有这两个都正常执行完了，才会执行某个任务。

• thenCombine 会将两个任务的执行结果作为方法入参，传递到指定方法中，且有返回值

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)

• thenAcceptBoth 会将两个任务的执行结果作为方法入参，传递到指定方法中，且无返回值

public <U> CompletableFuture<Void> thenAcceptBoth(
        CompletionStage<? extends U> other,
        BiConsumer<? super T, ? super U> action)

public <U> CompletableFuture<Void> thenAcceptBothAsync(
        CompletionStage<? extends U> other,
        BiConsumer<? super T, ? super U> action)

public <U> CompletableFuture<Void> thenAcceptBothAsync(
        CompletionStage<? extends U> other,
        BiConsumer<? super T, ? super U> action, Executor executor)

• runAfterBoth 不会把执行结果当做方法入参，且没有返回值。

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)

2 OR

将两个CompletableFuture组合起来，只要其中一个执行完了,就会执行某个任务。

• applyToEither：会将已经执行完成的任务，作为方法入参，传递到指定方法中，且有返回值

public <U> CompletableFuture<U> applyToEither(
        CompletionStage<? extends T> other, Function<? super T, U> fn)

public <U> CompletableFuture<U> applyToEitherAsync(
        CompletionStage<? extends T> other, Function<? super T, U> fn) 

public <U> CompletableFuture<U> applyToEitherAsync(
        CompletionStage<? extends T> other, Function<? super T, U> fn,
        Executor executor)

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

• runAfterEither： 不会把执行结果当做方法入参，且没有返回值。

3 AllOf

所有任务都执行完成后，才执行 allOf返回的CompletableFuture。如果任意一个任务异常，allOf的CompletableFuture，执行get方法，会抛出异常

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

4 AnyOf

任意一个任务执行完，就执行anyOf返回的CompletableFuture。如果执行的任务异常，anyOf的CompletableFuture，执行get方法，会抛出异常

public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

5 thenCompose

thenCompose方法会在某个任务执行完成后，将该任务的执行结果,作为方法入参,去执行指定的方法。该方法会返回一个新的CompletableFuture实例

public <U> CompletableFuture<U> thenCompose(
        Function<? super T, ? extends CompletionStage<U>> fn)

public <U> CompletableFuture<U> thenComposeAsync(
        Function<? super T, ? extends CompletionStage<U>> fn)

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

refer https://juejin.cn/post/6970558076642394142