CompletableFuture多任务组合回调

发布时间 2023-08-10 21:47:23作者: TestRookie

 

1、AND组合关系

thenCombine / thenAcceptBoth / runAfterBoth都表示:「当任务一和任务二都完成再执行任务三」

区别在于:

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

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

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

public void testCompletableThenCombine() throws ExecutionException, InterruptedException {  
    //创建线程池  
    ExecutorService executorService = Executors.newFixedThreadPool(10);  
    //开启异步任务1  
    CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {  
        System.out.println("异步任务1,当前线程是:" + Thread.currentThread().getId());  
        int result = 1 + 1;  
        System.out.println("异步任务1结束");  
        return result;  
    }, executorService);  

    //开启异步任务2  
    CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {  
        System.out.println("异步任务2,当前线程是:" + Thread.currentThread().getId());  
        int result = 1 + 1;  
        System.out.println("异步任务2结束");  
        return result;  
    }, executorService);  

    //任务组合  
    CompletableFuture<Integer> task3 = task.thenCombineAsync(task2, (f1, f2) -> {  
        System.out.println("执行任务3,当前线程是:" + Thread.currentThread().getId());  
        System.out.println("任务1返回值:" + f1);  
        System.out.println("任务2返回值:" + f2);  
        return f1 + f2;  
    }, executorService);  

    Integer res = task3.get();  
    System.out.println("最终结果:" + res);  
}  

运行结果

异步任务1,当前线程是:17  
异步任务1结束  
异步任务2,当前线程是:18  
异步任务2结束  
执行任务3,当前线程是:19  
任务1返回值:2  
任务2返回值:2  
最终结果:4  

2、OR组合关系

applyToEither / acceptEither / runAfterEither 都表示:「两个任务,只要有一个任务完成,就执行任务三」

区别在于:

  • 「runAfterEither」:不会把执行结果当做方法入参,且没有返回值

  • 「acceptEither」: 会将已经执行完成的任务,作为方法入参,传递到指定方法中,且无返回值

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

public void testCompletableEitherAsync() {  
    //创建线程池  
    ExecutorService executorService = Executors.newFixedThreadPool(10);  
    //开启异步任务1  
    CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {  
        System.out.println("异步任务1,当前线程是:" + Thread.currentThread().getId());  

        int result = 1 + 1;  
        System.out.println("异步任务1结束");  
        return result;  
    }, executorService);  

    //开启异步任务2  
    CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {  
        System.out.println("异步任务2,当前线程是:" + Thread.currentThread().getId());  
        int result = 1 + 2;  
        try {  
            Thread.sleep(3000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
        System.out.println("异步任务2结束");  
        return result;  
    }, executorService);  

    //任务组合  
    task.acceptEitherAsync(task2, (res) -> {  
        System.out.println("执行任务3,当前线程是:" + Thread.currentThread().getId());  
        System.out.println("上一个任务的结果为:"+res);  
    }, executorService);  
}  

运行结果

//通过结果可以看出,异步任务2都没有执行结束,任务3获取的也是1的执行结果  
异步任务1,当前线程是:17  
异步任务1结束  
异步任务2,当前线程是:18  
执行任务3,当前线程是:19  
上一个任务的结果为:2  

注意

如果把上面的核心线程数改为1也就是

ExecutorService executorService = Executors.newFixedThreadPool(1);  
 

运行结果就是下面的了,会发现根本没有执行任务3,显然是任务3直接被丢弃了。

异步任务1,当前线程是:17  
异步任务1结束  
异步任务2,当前线程是:17  

3、多任务组合

  • 「allOf」:等待所有任务完成

  • 「anyOf」:只要有一个任务完成

示例

allOf:等待所有任务完成

public void testCompletableAallOf() throws ExecutionException, InterruptedException {  
    //创建线程池  
    ExecutorService executorService = Executors.newFixedThreadPool(10);  
    //开启异步任务1  
    CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {  
        System.out.println("异步任务1,当前线程是:" + Thread.currentThread().getId());  
        int result = 1 + 1;  
        System.out.println("异步任务1结束");  
        return result;  
    }, executorService);  

    //开启异步任务2  
    CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {  
        System.out.println("异步任务2,当前线程是:" + Thread.currentThread().getId());  
        int result = 1 + 2;  
        try {  
            Thread.sleep(3000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
        System.out.println("异步任务2结束");  
        return result;  
    }, executorService);  

    //开启异步任务3  
    CompletableFuture<Integer> task3 = CompletableFuture.supplyAsync(() -> {  
        System.out.println("异步任务3,当前线程是:" + Thread.currentThread().getId());  
        int result = 1 + 3;  
        try {  
            Thread.sleep(4000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
        System.out.println("异步任务3结束");  
        return result;  
    }, executorService);  

    //任务组合  
    CompletableFuture<Void> allOf = CompletableFuture.allOf(task, task2, task3);  

    //等待所有任务完成  
    allOf.get();  
    //获取任务的返回结果  
    System.out.println("task结果为:" + task.get());  
    System.out.println("task2结果为:" + task2.get());  
    System.out.println("task3结果为:" + task3.get());  
}  

anyOf: 只要有一个任务完成

 
public void testCompletableAnyOf() throws ExecutionException, InterruptedException {  
    //创建线程池  
    ExecutorService executorService = Executors.newFixedThreadPool(10);  
    //开启异步任务1  
    CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {  
        int result = 1 + 1;  
        return result;  
    }, executorService);  

    //开启异步任务2  
    CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {  
        int result = 1 + 2;  
        return result;  
    }, executorService);  

    //开启异步任务3  
    CompletableFuture<Integer> task3 = CompletableFuture.supplyAsync(() -> {  
        int result = 1 + 3;  
        return result;  
    }, executorService);  

    //任务组合  
    CompletableFuture<Object> anyOf = CompletableFuture.anyOf(task, task2, task3);  
    //只要有一个有任务完成  
    Object o = anyOf.get();  
    System.out.println("完成的任务的结果:" + o);  
}  

CompletableFuture使用有哪些注意点

 

使用的一些注意点。

1、Future需要获取返回值,才能获取异常信息

@Test  
public void testWhenCompleteExceptionally() {  
    CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {  
        if (1 == 1) {  
            throw new RuntimeException("出错了");  
        }  
        return 0.11;  
    });  

    //如果不加 get()方法这一行,看不到异常信息  
    //future.get();  
}  

Future需要获取返回值,才能获取到异常信息。如果不加 get()/join()方法,看不到异常信息。


2、CompletableFuture的get()方法是阻塞的

CompletableFutureget()方法是阻塞的,如果使用它来获取异步调用的返回值,需要添加超时时间。

//反例  
 CompletableFuture.get();  
//正例  
CompletableFuture.get(5, TimeUnit.SECONDS);  

3、不建议使用默认线程池

CompletableFuture代码中又使用了默认的 「ForkJoin线程池」,处理的线程个数是电脑 「CPU核数-1」。在大量请求过来的时候,处理逻辑复杂的话,响应会很慢。一般建议使用自定义线程池,优化线程池配置参数。

4、自定义线程池时,注意饱和策略

CompletableFuture的get()方法是阻塞的,我们一般建议使用future.get(5, TimeUnit.SECONDS)。并且一般建议使用自定义线程池。

但是如果线程池拒绝策略是DiscardPolicy或者DiscardOldestPolicy,当线程池饱和时,会直接丢弃任务,不会抛弃异常。因此建议,CompletableFuture线程池策略最好使用AbortPolicy,然后耗时的异步线程,做好线程池隔离哈。