【Java8新特性---＞异步处理】CompletableFuture

小咪咪 2023-10-10 22:00 27阅读 0赞

一、引入  
假设一个商品详情页需要以下操作：

查询展示商品的基本信息耗时：0.5s  
查询展示商品的销售信息耗时：0.7s  
查询展示商品的图片信息耗时：1s  
查询展示商品销售属性耗时：0.3s  
查询展示商品规格属性耗时：1.5s  
查询展示商品详情信息耗时：1s  
即使每个查询时间耗时不多，但是加起来却需要很长耗时。为了减少线性执行造成耗时的累积，这就需要引入异步处理做优化。

二、Future介绍  
Future是Java 5添加的类，用来描述一个异步计算的结果。

优点：

**可以使用 isDone 方法检查计算是否完成。  
使用 get 阻塞住调用线程，直到计算完成返回结果。**  
可以使用 cancel 方法停止任务的执行。  
缺点：

对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。  
**阻塞的方式与我们想及时得到计算结果的期望相违背。  
轮询的方式会消耗大量CPU资源，并且不能及时得到计算结果。**  
功能：

boolean cancel(boolean mayInterruptIfRunning);

> 1、尝试取消执行此任务。如果任务已经完成、已被取消或由于其他原因无法取消，则此尝试将失败。如果成功，并且调用 cancel  
> 时此任务尚未启动，则此任务永远不会运行。  
> 2、 如果任务已经开始，则mayInterruptIfRunning  
> 参数确定是否应中断执行此任务的线程以尝试停止任务。 参数mayInterruptIfRunning 为true，表示执行此任务的线程应该被中断；否则，允许进行中的任务完成。 3、此方法返回后，后续调用isDone 将始终返回true。  
> 4、如果此方法返回true，则对isCancelled 的后续调用将始终返回true。

boolean isCancelled();

> 如果此任务在正常完成之前被取消，则返回true，否则返回false。

boolean isDone();

> 如果此任务完成，则返回 true。任务完成可能是由于正常终止、异常或取消——在所有这些情况下，此方法将返回 true。

V get() throws InterruptedException, ExecutionException;

> 阻塞直至任务完成，然后检索其结果。 throws CancellationException：如果计算被取消 throws ExecutionException：如果计算抛出异常 throws InterruptedException：如果当前线程在等待时被中断
> 
> V get(long timeout, TimeUnit unit) throws InterruptedException,  
> ExecutionException, TimeoutException;  
> 如有必要，最多等待计算完成的给定时间，然后检索其结果（如果可用）。  
> 参数timeout：等待的最长时间 参数unit：超时参数的时间单位  
> throws CancellationException：如果计算被取消 throws  
> ExecutionException：如果计算抛出异常 throws InterruptedException：如果当前线程在等待时被中断  
> throws TimeoutException：如果等待超时

三、CompletableFuture  
1、介绍  
在Java 8中, 新增加了一个包含50个方法左右的类：CompletableFuture，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

**CompletableFuture类实现了Future接口，所以你还是可以像以前一样通过 get方法阻塞或者轮询的方式获得结果，但是这种方式不推荐使用。**

2、创建异步对象（runAsync、supplyAsync）  
CompletableFuture 提供了四个静态方法来创建一个异步操作。  
（带有Async默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。）

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 有返回结果**

测试代码：

public class CompletableFutureDemo {
 
 
 // corePoolSize:线程池的核心线程数量 线程池创建出来后就会 new Thread() 5个
 // maximumPoolSize:最大的线程数量，线程池支持的最大的线程数
 // keepAliveTime:存活时间，当线程数大于核心线程，空闲的线程的存活时间 50-5=45
 // unit:存活时间的单位
 // BlockingQueue<Runnable> workQueue:阻塞队列 当线程数超过了核心线程数据，那么新的请求到来的时候会加入到阻塞的队列中
 // new LinkedBlockingQueue<>() 默认队列的长度是 Integer.MAX 那这个就太大了，所以我们需要指定队列的长度
 // threadFactory:创建线程的工厂对象
 // RejectedExecutionHandler handler:当线程数大于最大线程数的时候会执行的淘汰策略
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
 
 System.out.println("线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 50;
 System.out.println("线程结束了...");
 }, executor);
 System.out.println("main方法结束了…………");
 
 System.out.println("-----------------------------");
 CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 50;
 System.out.println("线程结束了...");
 return i;
 }, executor);
 System.out.println("integerCompletableFuture=" + integerCompletableFuture.get());
 }
 }

测试结果：  
![在这里插入图片描述][378e6e92c455491b8a212633215a94f3.png_pic_center]

3、whenCompleteAsync、exceptionally和handleAsync  
当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);
 
 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) ;

3.1、whenCompleteAsync  
可以获取异步任务的返回值和抛出的异常信息，但是不能修改返回结果。

测试代码：

public class CompletableFutureDemo1 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 2;
 System.out.println("线程结束了...");
 return i;
 }, executor).whenCompleteAsync((res, e) -> {
 // 不能修改返回值
 System.out.println("res= " + res);
 System.out.println("e=" + e);
 }, executor);
 System.out.println("main方法结束了…………");
 System.out.println("future=" + future.get());
 }
 }

测试结果：

![在这里插入图片描述][094977dcd548495d9972254fa2299d73.png_pic_center]

3.2、exceptionally  
当异步任务跑出了异常后会触发的方法，如果没有抛出异常该方法不会执行

测试代码：

public class CompletableFutureDemo1 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 0;
 System.out.println("线程结束了...");
 return i;
 }, executor).whenCompleteAsync((res, e) -> {
 // 不能修改返回值
 System.out.println("res= " + res);
 System.out.println("e=" + e);
 }, executor).exceptionally(e -> {
 
 System.out.println("exceptionally执行了，e = " + e);
 return 100;
 });
 System.out.println("main方法结束了…………");
 System.out.println("future=" + future.get());
 }
 }

测试结果： 主动触发算术异常

![在这里插入图片描述][1b4832f473ba47869e200ec3f46d6275.png_pic_center]

测试结果： 将int i = 100 / 0;改为int i = 100 / 5;

![在这里插入图片描述][14762f69ef2c4ab1a9a9ef69d5552fd5.png_pic_center]

3.2.1、拓展—>利用exceptionally达到显式地捕获相关异常的效果  
示例：

@Test
 public void test() {
 
 CompletableFuture<Integer> future = createNewFile();
 try {
 
 Integer flag = future.get();
 if (flag == 0) {
 
 System.out.println("创建成功！！！");
 } else if (flag == 1) {
 
 System.out.println("捕获到NullPointerException，创建失败！！！");
 } else if (flag == 2) {
 
 System.out.println("捕获到IOException，创建失败！！！");
 } else {
 
 System.out.println("捕获到其他异常，创建失败！！！");
 }
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 }
 
 private CompletableFuture<Integer> createNewFile() {
 
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
 
 File file = null;
 try {
 
 file.createNewFile();
 } catch (IOException e) {
 
 // 对于编译时异常，在CompletableFuture中不能直接向外抛出，
 throw new RuntimeException(e.getMessage());
 } catch (NullPointerException e) {
 
 throw e;
 }
 return 0;
 }).exceptionally(th -> {
 
 // 走到这里，说明创建失败
 if (th.getCause() instanceof NullPointerException) {
 
 // 显式地修改返回值
 return 1;
 } else if (th.getCause() instanceof RuntimeException) {
 
 return 2;
 }
 return 3;
 });
 return future;
 }

效果：  
![在这里插入图片描述][1c544b93eef3489ca7d82521e6a164a3.png_pic_center]

3.2.2、拓展—>completeExceptionally  
如果完成动作过程中抛出异常，将导致对get()和相关方法的调用引发给定的异常。

简而言之，future.completeExceptionally(e)是在完成过程中主动设置异常信息。（如果future未完成，抛出主动设置的异常信息。反之，则返回完成后的结果。）

案例：

@Test
 public void test03() throws InterruptedException {
 
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future start……");
 try {
 
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 int i = 100 / 20;
 System.out.println("future end……");
 return i;
 });
 Thread.sleep(3 * 1000);
 future.completeExceptionally(new RuntimeException("未完成时主动抛出异常！！！"));
 try {
 
 Integer res = future.get();
 System.out.println("res = " + res);
 } catch (ExecutionException e) {
 
 // 可捕获主动抛出的异常
 System.out.println("捕获主动抛出的异常ExecutionException");
 e.printStackTrace();
 }
 }

效果 （future已完成计算）：

![在这里插入图片描述][bde11d6f9c3548048f2c188a8ba22083.png_pic_center]

如果将主线程Thread.sleep(3 \* 1000);改为Thread.sleep(1 \* 1000);  
结果如下 （future未完成计算）：  
![在这里插入图片描述][45aeb49970b742e2954127fb3455149c.png_pic_center]

3.2.3、拓展—>obtrudeException  
强制导致方法get()和相关方法的后续调用抛出给定的异常，无论是否已完成。

3.3、handleAsync  
可以获取异步任务的返回值和抛出的异常信息，而且可以显示地修改返回的结果

测试代码：

public class CompletableFutureDemo1 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 System.out.println("线程结束了...");
 return i;
 }, executor).handle((res, e) -> {
 
 System.out.println("res = " + res);
 System.out.println("e = " + e);
 return 200;
 });
 System.out.println("main方法结束了…………");
 System.out.println("future=" + future.get());
 }
 }

测试效果：

![在这里插入图片描述][af7c63df507046e8b7b338b91c825db2.png_pic_center]

4、线程串行方法  
thenRunAsync 方法：只要之前的执行完成就执行 thenRun的后续操作。（无接受参数，无返回）  
thenAcceptAsync 方法：消费者模式，接受上一个任务处理的结果，并消费处理，无返回结果  
thenApplyAsync 方法：当一个线程依赖另一个线程，获取上一个任务的返回结果，并返回当前任务的结果。  
（带有Async默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。）

public CompletionStage<Void> thenRun(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action,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 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);

4.1、thenRunAsync 实现代码

public class CompletableFutureDemo2 {
        
    
        private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                50,
                10,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque(1000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
        
            System.out.println("main方法开始了…………");
            CompletableFuture.runAsync(() -> {
        
                System.out.println("线程开始了...");
                System.out.println("当前线程---->" + Thread.currentThread().getName());
                int i = 100 / 5;
                System.out.println("线程结束了...");
            }, executor).thenRunAsync(()->{
        
                System.out.println("thenRunAsync我进行操作了……………………");
            });
        }
        
    }

4.2、thenRunAsync 实现效果  
![在这里插入图片描述][12fa17107b5a4c4cab69537539b8a8d7.png_pic_center]

4.3、thenAcceptAsync 实现代码

public class CompletableFutureDemo2 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 System.out.println("线程结束了...");
 return i;
 }, executor).thenAcceptAsync(res -> {
 
 System.out.println("thenAcceptAsync-------->" + res);
 });
 }
 
 }

4.4、thenAcceptAsync 实现效果  
![在这里插入图片描述][0c88f9e4e4b1471d94c5c5cf5b53ddd1.png_pic_center]

4.5、thenApplyAsync 实现代码

public class CompletableFutureDemo2 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 System.out.println("线程结束了...");
 return i;
 }, executor).thenApplyAsync(res -> {
 
 System.out.println("thenAcceptAsync-------->" + res);
 return 200;
 });
 
 System.out.println("future--------->" + future.get());
 }
 
 }

4.6、thenApplyAsync 实现效果  
![在这里插入图片描述][1392dd34a6d540f5a7882bd5f7b79fa9.png_pic_center]

5、等待两个任务执行完成后才会触发  
runAfterBothAsync 方法：不可以获取前面两线程的返回结果，本身也没有返回结果。  
thenAcceptBothAsync 方法：可以获取前面两线程的返回结果，本身没有返回结果。  
thenCombineAsync 方法：可以获取前面两线程的返回结果，本身也有返回结果  
（带有Async默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。）

5.1、runAfterBothAsync 实现代码

public class CompletableFutureDemo3 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 // try {
 
 // Thread.sleep(3000);
 // } catch (InterruptedException e) {
 
 // e.printStackTrace();
 // }
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 future1.runAfterBothAsync(future2,()->{
 
 System.out.println("任务3执行了");
 },executor);
 }
 
 }

5.2、runAfterBothAsync 实现效果  
如果放开Thread.sleep(3000);，那么 runAfterBothAsync 中的代码会等3s后（也就是等待 future1和 future2都执行完）才执行。  
![在这里插入图片描述][0fa7bfb52e124389b5f866647ea13fda.png_pic_center]

5.3、thenAcceptBothAsync 实现代码

public class CompletableFutureDemo3 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 // try {
 
 // Thread.sleep(3000);
 // } catch (InterruptedException e) {
 
 // e.printStackTrace();
 // }
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 future1.thenAcceptBothAsync(future2, (res1, res2) -> {
 
 System.out.println("thenAcceptBothAsync开始了");
 System.out.println("res1 = " + res1);
 System.out.println("res2 = " + res2);
 System.out.println("thenAcceptBothAsync结束了");
 },executor);
 }
 
 }

5.4、thenAcceptBothAsync 实现效果  
![在这里插入图片描述][84863234b6784a1e9c6c98903a471c08.png_pic_center]

5.5、thenCombineAsync 实现代码

public class CompletableFutureDemo3 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 // try {
 
 // Thread.sleep(3000);
 // } catch (InterruptedException e) {
 
 // e.printStackTrace();
 // }
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future3 = future1.thenCombineAsync(future2, (res1, res2) -> {
 
 System.out.println("future3开始了");
 return res1 + res2;
 }, executor);
 System.out.println("future3.get()=" + future3.get());
 }
 
 }

5.6、thenCombineAsync 实现效果  
![在这里插入图片描述][a868dbb952914b71b55cf04a3c3ddf9e.png_pic_center]

6、两个任务完成一个就会触发  
两个任务只要有一个完成就会触发。  
（对于acceptEitherAsync、applyToEitherAsync可接受前面两任务返回结果来说，如果任务有返回值，哪个任务先执行完先获取其结果作为参数）

> runAfterEitherAsync 方法：  
> 不可以获取前面两线程的返回结果，本身也没有返回结果。  
> acceptEitherAsync方法：  
> 可以获取前面两线程的返回结果，本身也没有返回结果。 applyToEitherAsync方法：  
> 可以获取前面两线程的返回结果，本身有返回结果。 （带有Async默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。）

6.1、runAfterEitherAsync 实现代码

public class CompletableFutureDemo4 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 // try {
 
 // Thread.sleep(3000);
 // } catch (InterruptedException e) {
 
 // e.printStackTrace();
 // }
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 // try {
 
 // Thread.sleep(3000);
 // } catch (InterruptedException e) {
 
 // e.printStackTrace();
 // }
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 future1.runAfterEitherAsync(future2,()->{
 
 System.out.println("runAfterEitherAsync任务执行了");
 },executor);
 
 }
 
 }

6.2、runAfterEitherAsync 实现效果  
![在这里插入图片描述][97bb6f65b5f0493fa405336e288f552a.png_pic_center]

6.3、acceptEitherAsync 实现代码

public class CompletableFutureDemo4 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 try {
 
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 // try {
 
 // Thread.sleep(3000);
 // } catch (InterruptedException e) {
 
 // e.printStackTrace();
 // }
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 future1.acceptEitherAsync(future2, (res) -> {
 
 System.out.println("acceptEitherAsync开始了");
 System.out.println("res = " + res);
 },executor);
 
 }
 
 }

6.4、acceptEitherAsync 实现效果  
![在这里插入图片描述][b2a8f1bbba2a4fc2ade995666362bf51.png_pic_center]

6.5、applyToEitherAsync 实现代码

public class CompletableFutureDemo4 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 try {
 
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 // try {
 
 // Thread.sleep(3000);
 // } catch (InterruptedException e) {
 
 // e.printStackTrace();
 // }
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future3 = future1.applyToEitherAsync(future2, (res) -> {
 
 System.out.println("future3开始了");
 System.out.println("res = " + res);
 return res;
 }, executor);
 System.out.println("future3.get()=" + future3.get());
 }
 
 }

6.6、applyToEitherAsync 实现效果  
![在这里插入图片描述][382fb2bf918544d09481ed95df8d050a.png_pic_center]

7、多任务组合  
anyOf 方法：只要有一个任务完成。  
allOf 方法：等待所有任务完成。

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

7.1、anyOf 实现代码  
只要有一个任务完成就会触发。

public class CompletableFutureDemo5 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 /**
 * anyOf：只要有一个线程完成，那么就不阻塞
 * allOf：所有线程都完成，在 get方法阻塞直至所有线程都完成
 */
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 try {
 
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2);
 anyOf.get();
 System.out.println("主任务完成anyOf：" + anyOf.get());
 }
 
 }

7.2、anyOf 实现效果

![在这里插入图片描述][86d9214f91bc4e7296d3036eef61a80a.png_pic_center]

7.3、allOf 实现代码  
阻塞等待所有任务完成才会触发。

public class CompletableFutureDemo5 {
 
 
 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
 5,
 50,
 10,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque(1000),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
 );
 
 /**
 * anyOf：只要有一个线程完成，那么就不阻塞
 * allOf：所有线程都完成，在 get方法阻塞直至所有线程都完成
 */
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 System.out.println("main方法开始了…………");
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future1线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 5;
 try {
 
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("future1线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2线程开始了...");
 System.out.println("当前线程---->" + Thread.currentThread().getName());
 int i = 100 / 20;
 System.out.println("future2线程结束了...");
 return i;
 }, executor);
 
 CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2);
 allOf.get(); // 阻塞在这个位置，等待所有线程的完成
 System.out.println("主任务完成allOf：" + future1.get() + "，" + future2.get());
 }
 
 }

7.4、allOf 实现效果  
![在这里插入图片描述][a4d50e950f7349f5859357acc42749d5.png_pic_center]

8、细节  
8.1、thenApply与thenCompose区别

public CompletableFuture thenApply(
 Function<? super T,? extends U> fn) {
 
 return uniApplyStage(null, fn);
 }
 public CompletableFuture thenCompose(
 Function<? super T, ? extends CompletionStage> fn) {
 
 return uniComposeStage(null, fn);
 }

thenApply：返回的是泛型中的类型转化为返回值类型的CompletableFuture对象。  
thenCompose：返回的是一个扁平化的CompletableFuture对象。  
（用来连接两个CompletableFuture，是生成一个新的CompletableFuture。特别像stream().flatMap扁平化处理）  
区别：当返回是CompletableFuture的话，thenApply是嵌套，而thenCompose扁平化。

public class CompletableFutureTest {
 
 
 public static void main(String[] args) {
 
 CompletableFuture<Integer> apply = CompletableFuture.supplyAsync(() -> {
 
 int i;
 try {
 
 i = 100 / 2;
 } catch (Exception e) {
 
 throw new RuntimeException("算术异常！！！");
 }
 return i;
 });
 // thenApply返回的如果是CompletableFuture，就会嵌套起来
 CompletableFuture<? extends CompletableFuture<?>> apply1 = apply.thenApply(res -> {
 
 System.out.println("res1 = " + res);
 System.out.println("apply1当前还在执行中……");
 return CompletableFuture.completedFuture(res);
 });
 CompletableFuture<? extends CompletableFuture<?>> apply2 = apply.thenApply(res -> {
 
 System.out.println("res2 = " + res);
 System.out.println("apply2当前还在执行中……");
 return CompletableFuture.completedFuture(res);
 });
 // thenCompose不同，返回的扁平化之后的一维CompletableFuture
 CompletableFuture<?> compose = apply.thenCompose(res -> {
 
 System.out.println("res3 = " + res);
 System.out.println("compose当前还在执行中……");
 return CompletableFuture.completedFuture(res);
 });
 
 System.out.println(apply1 == apply2);
 try {
 
 apply.get();
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 }
 }

测试效果：  
![在这里插入图片描述][6860a34c8ec846058c1defae81d4df45.png_pic_center]

**总结**  
![在这里插入图片描述][8f7c7db062664e76872ffbb3d5a9ff40.png_pic_center]

[378e6e92c455491b8a212633215a94f3.png_pic_center]: https://img-blog.csdnimg.cn/378e6e92c455491b8a212633215a94f3.png#pic_center
[094977dcd548495d9972254fa2299d73.png_pic_center]: https://img-blog.csdnimg.cn/094977dcd548495d9972254fa2299d73.png#pic_center
[1b4832f473ba47869e200ec3f46d6275.png_pic_center]: https://img-blog.csdnimg.cn/1b4832f473ba47869e200ec3f46d6275.png#pic_center
[14762f69ef2c4ab1a9a9ef69d5552fd5.png_pic_center]: https://img-blog.csdnimg.cn/14762f69ef2c4ab1a9a9ef69d5552fd5.png#pic_center
[1c544b93eef3489ca7d82521e6a164a3.png_pic_center]: https://img-blog.csdnimg.cn/1c544b93eef3489ca7d82521e6a164a3.png#pic_center
[bde11d6f9c3548048f2c188a8ba22083.png_pic_center]: https://img-blog.csdnimg.cn/bde11d6f9c3548048f2c188a8ba22083.png#pic_center
[45aeb49970b742e2954127fb3455149c.png_pic_center]: https://img-blog.csdnimg.cn/45aeb49970b742e2954127fb3455149c.png#pic_center
[af7c63df507046e8b7b338b91c825db2.png_pic_center]: https://img-blog.csdnimg.cn/af7c63df507046e8b7b338b91c825db2.png#pic_center
[12fa17107b5a4c4cab69537539b8a8d7.png_pic_center]: https://img-blog.csdnimg.cn/12fa17107b5a4c4cab69537539b8a8d7.png#pic_center
[0c88f9e4e4b1471d94c5c5cf5b53ddd1.png_pic_center]: https://img-blog.csdnimg.cn/0c88f9e4e4b1471d94c5c5cf5b53ddd1.png#pic_center
[1392dd34a6d540f5a7882bd5f7b79fa9.png_pic_center]: https://img-blog.csdnimg.cn/1392dd34a6d540f5a7882bd5f7b79fa9.png#pic_center
[0fa7bfb52e124389b5f866647ea13fda.png_pic_center]: https://img-blog.csdnimg.cn/0fa7bfb52e124389b5f866647ea13fda.png#pic_center
[84863234b6784a1e9c6c98903a471c08.png_pic_center]: https://img-blog.csdnimg.cn/84863234b6784a1e9c6c98903a471c08.png#pic_center
[a868dbb952914b71b55cf04a3c3ddf9e.png_pic_center]: https://img-blog.csdnimg.cn/a868dbb952914b71b55cf04a3c3ddf9e.png#pic_center
[97bb6f65b5f0493fa405336e288f552a.png_pic_center]: https://img-blog.csdnimg.cn/97bb6f65b5f0493fa405336e288f552a.png#pic_center
[b2a8f1bbba2a4fc2ade995666362bf51.png_pic_center]: https://img-blog.csdnimg.cn/b2a8f1bbba2a4fc2ade995666362bf51.png#pic_center
[382fb2bf918544d09481ed95df8d050a.png_pic_center]: https://img-blog.csdnimg.cn/382fb2bf918544d09481ed95df8d050a.png#pic_center
[86d9214f91bc4e7296d3036eef61a80a.png_pic_center]: https://img-blog.csdnimg.cn/86d9214f91bc4e7296d3036eef61a80a.png#pic_center
[a4d50e950f7349f5859357acc42749d5.png_pic_center]: https://img-blog.csdnimg.cn/a4d50e950f7349f5859357acc42749d5.png#pic_center
[6860a34c8ec846058c1defae81d4df45.png_pic_center]: https://img-blog.csdnimg.cn/6860a34c8ec846058c1defae81d4df45.png#pic_center
[8f7c7db062664e76872ffbb3d5a9ff40.png_pic_center]: https://img-blog.csdnimg.cn/8f7c7db062664e76872ffbb3d5a9ff40.png#pic_center