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

news/2024/12/5 5:31:22/

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

查询展示商品的基本信息耗时：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 <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> 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());}
}

测试结果：
在这里插入图片描述

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

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

测试结果：

在这里插入图片描述

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

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

在这里插入图片描述

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

在这里插入图片描述

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

效果：
在这里插入图片描述

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

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

案例：

 @Testpublic 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已完成计算）：

在这里插入图片描述

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

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

测试效果：

在这里插入图片描述

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 <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.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 实现效果
在这里插入图片描述

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 实现效果
在这里插入图片描述

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 实现效果
在这里插入图片描述

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都执行完）才执行。
在这里插入图片描述

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 实现效果
在这里插入图片描述

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 实现效果
在这里插入图片描述

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 实现效果
在这里插入图片描述

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 实现效果
在这里插入图片描述

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 实现效果
在这里插入图片描述

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 实现效果

在这里插入图片描述

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 实现效果
在这里插入图片描述

8、细节
8.1、thenApply与thenCompose区别

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) {return uniApplyStage(null, fn);
}
public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> 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不同，返回的扁平化之后的一维CompletableFutureCompletableFuture<?> 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();}}
}

测试效果：
在这里插入图片描述

总结
在这里插入图片描述


http://www.ppmy.cn/news/87424.html

相关文章

实验11 人工神经网络(2)

实验11 人工神经网络(2)

1. 实验目的 ①掌握梯度下降法的优化算法&#xff1b; ②能够使用tf.keras构建Sequential模型&#xff0c;完成多分类任务。 2. 实验内容 ①下载MNIST数据集&#xff0c;建立神经网络模型&#xff0c;实现对MNIST手写数字数据集的识别&#xff0c;调整超参数和训练参数&…
阅读更多...
一则历史：为什么网络路径前加一个盘符还能正常工作

一则历史：为什么网络路径前加一个盘符还能正常工作

有一个比较知名的奇异特性&#xff1a;文件系统在解析 UNC(Universal Naming Convention) 路径时&#xff0c;会故意忽略掉最前面添加的盘符字母。 举个例子&#xff0c;假设服务器上有一个共享文件夹&#xff0c;其路径为&#xff1a;\\server\share\directory&#xff0c;如果…
阅读更多...
android 12.0 设置app为默认浏览器

android 12.0 设置app为默认浏览器

1.概述 在12.0的产品定制化中，如果系统安装多个浏览器时，需要设置默认浏览器来完成需求，这就需要看系统设置中的相关源码 当出现多个浏览器时，该如何设置默认浏览器呢, 其实在Settings 默认应用->浏览器应用 当点击选择浏览器时会调用 /package/app/PermissionControll…
阅读更多...
基于html+css的图片展示93

基于html+css的图片展示93

准备项目 项目开发工具 Visual Studio Code 1.44.2 版本: 1.44.2 提交: ff915844119ce9485abfe8aa9076ec76b5300ddd 日期: 2020-04-16T16:36:23.138Z Electron: 7.1.11 Chrome: 78.0.3904.130 Node.js: 12.8.1 V8: 7.8.279.23-electron.0 OS: Windows_NT x64 10.0.19044 项目…
阅读更多...
node.js+vue房屋租赁管理系统z0g8w

node.js+vue房屋租赁管理系统z0g8w

本系统主要包括以下功能模块&#xff1a;租户、出租人、房源信息、预约看房、合同信息等模块。 其中设计的主要功能如下&#xff1a; &#xff08;1&#xff09;用户的注册和登录本系统&#xff0c;登录到系统的首页。 &#xff08;2&#xff09;用户可以发布自己的房源信息…
阅读更多...
C++的指针和引用

C++的指针和引用

文章目录 C的指针和引用C指针C中内存单元内容和地址指针的定义和间接访问操作指针和数组左值和右值 几种C中的原始指针原始指针的基本运算 存储区域划分栈和队列代码在内存单元中的分布cpp动态分配资源和回收原则资源管理方案-RAIIC中几种变量对比 内存泄漏智能指针C的智能指针…
阅读更多...
【Nginx】- 02 Nginx反向代理、负载均衡、动静分离、虚拟域名配置

【Nginx】- 02 Nginx反向代理、负载均衡、动静分离、虚拟域名配置

Nginx反向代理、负载均衡、动静分离、虚拟域名配置 1 反向代理1.1 nginx配置 2 负载均衡2.1 ngxin配置2.2 nginx负载均衡策略2.2.1 默认轮询2.2.2 权重2.2.3 iphash2.2.4 least_conn 3 动静分离3.1 打包前端项目3.2 部署前端项目3.3 配置静态图片资源访问 4 虚拟域名配置4.1 修…
阅读更多...
RLHF中的PPO算法原理及其实现

RLHF中的PPO算法原理及其实现

RLHF中的PPO算法原理及其实现 ChatGPT是基于InstructGPT实现的多轮对话生成式大模型。ChatGPT主要涉及到的技术包括&#xff1a; 指令微调&#xff08;Instruction-tuning&#xff09;&#xff1b;因果推断&#xff08;Causal Language Modeling&#xff09;&#xff1b;人类…
阅读更多...
最新文章

Copyright @ 2022~2023