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

article/2025/10/28 7:18:59

一、引入

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

  • 查询展示商品的基本信息耗时：0.5s
  • 查询展示商品的销售信息耗时：0.7s
  • 查询展示商品的图片信息耗时：1s
  • 查询展示商品销售属性耗时：0.3s
  • 查询展示商品规格属性耗时：1.5s
  • 查询展示商品详情信息耗时：1s

即使每个查询时间耗时不多，但是加起来却需要很长耗时。为了减少线性执行造成耗时的累积，这就需要引入异步处理做优化。

二、Future介绍

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

优点：

  1. 可以使用 isDone 方法检查计算是否完成。
  2. 使用 get 阻塞住调用线程，直到计算完成返回结果。
  3. 可以使用 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是扁平化的。

在这里插入图片描述


http://chatgpt.dhexx.cn/article/ixKgAeay.shtml

相关文章

CompletableFuture API

CompletableFuture API

目录 1. 为什么要用CompletableFuture1.1 API 2. CompletableFuture Demo1. 创建CompletableFuture2. 定义CompletableFuture完成时和异常时需要回调的实例3. CompletableFuture的优点 3. demo&#xff1a;多个CompletableFuture串行执行4. demo&#xff1a;多个CompletableFut…
阅读更多...
CompletableFuture实战与分析

CompletableFuture实战与分析

Future对于结果的获取不够好&#xff0c;只能通过阻塞或者轮询的方式得到任务的结果。在Java8中Doug Lea大师提供了一个CompletableFuture工具类&#xff0c;可以更优雅的对异步并行操作进行编排。 Future VS CompletableFuture CompletableFuture支持手动完成任务&#xff0…
阅读更多...
Java8 CompletableFuture异步非阻塞做法

Java8 CompletableFuture异步非阻塞做法

创建异步任务 Future.submit supplyAsync / runAsync 异步回调 thenApply / thenApplyAsync thenAccept / thenRun exceptionally whenComplete handle 组合处理 thenCombine / thenAcceptBoth / runAfterBoth applyToEither / acceptEither / runAfterEither thenCom…
阅读更多...
2022.2.5 第十三次周报

2022.2.5 第十三次周报

文章目录 前言一、论文阅读《ROCKET: Exceptionally fast and accurate time series classification using random convolutional kernels》Abstract摘要Introduction介绍Method方法Kernels内核Transform转换Classifier分类器Complexity Analysis复杂性分析 Experiments实验Con…
阅读更多...
并发编程(十五)-CompletableFuture中常用方法的使用与分析

并发编程(十五)-CompletableFuture中常用方法的使用与分析

文章目录 一、CompletableFuture API介绍1. 描述2. CompletionStage3. CompletableFuture 4个核心静态方法&#xff08;1&#xff09;runAsync(Runnable runnable)&#xff08;2&#xff09;runAsync(Runnable runnable, Executor executor)&#xff08;3&#xff09;supplyAsy…
阅读更多...
Java 编程问题：十一、并发-深入探索

Java 编程问题：十一、并发-深入探索

原文&#xff1a;Java Coding Problems 协议&#xff1a;CC BY-NC-SA 4.0 贡献者&#xff1a;飞龙 本文来自【ApacheCN Java 译文集】&#xff0c;自豪地采用谷歌翻译。 本章包括涉及 Java 并发的 13 个问题&#xff0c;涉及 Fork/Join 框架、CompletableFuture、ReentrantLock…
阅读更多...
线程(十二)---CompletableFuture（三）

线程(十二)---CompletableFuture（三）

写在前面&#xff1a;各位看到此博客的小伙伴&#xff0c;如有不对的地方请及时通过私信我或者评论此博客的方式指出&#xff0c;以免误人子弟。多谢&#xff01; 示例五&#xff1a;异常处理 接着上一篇记录一下CompletableFuture的异常处理&#xff0c;异常处理通常使用…
阅读更多...
dice loss

dice loss

Dice Loss 最先是在VNet 这篇文章中被提出&#xff0c;后来被广泛的应用在了医学影像分割之中。 Dice 系数 Dice系数作为损失函数的原因和混淆矩阵有着很大的关系&#xff0c;下图给出的是一个混淆矩阵&#xff1a; 其中的一些关键指标如下&#xff1a; 精确率(precision)表…
阅读更多...
Hinge loss

Hinge loss

声明&#xff1a; 参考自维基百科后面可能会更新 Hinge Loss 在机器学习中&#xff0c;hinge loss作为一个损失函数(loss function)&#xff0c;通常被用于最大间隔算法(maximum-margin)&#xff0c;而最大间隔算法又是SVM(支持向量机support vector machines)用到的重要算法…
阅读更多...
【深度学习】一文读懂机器学习常用损失函数（Loss Function）

【深度学习】一文读懂机器学习常用损失函数（Loss Function）

【深度学习】一文读懂机器学习常用损失函数&#xff08;Loss Function&#xff09; 最近太忙已经好久没有写博客了&#xff0c;今天整理分享一篇关于损失函数的文章吧&#xff0c;以前对损失函数的理解不够深入&#xff0c;没有真正理解每个损失函数的特点以及应用范围&#x…
阅读更多...
Pytorch之loss（损失函数）

Pytorch之loss（损失函数）

损失函数也在torch.nn下&#xff0c;具体可以参考文档&#xff0c;也可以参考官网 先根据L1Loss举例 我个人感觉这里的描述还是到官网的文档找比较好&#xff0c;公式看的比文档清楚 import torch from torch import nninputs torch.tensor([[3, 2, 1],[1, 2, 3]], dtypetorch…
阅读更多...
机器学习 损失函数 Loss function

机器学习 损失函数 Loss function

损失函数 最小二乘法极大似然估计法交叉熵 【本文根据B站-王木头学科学-视频所学】 在梯度下降中&#xff0c;所求的梯度其实就是损失函数的梯度。 损失函数有三种设计方法&#xff1a; &#xff08;1&#xff09;最小二乘法 &#xff08;2&#xff09;极大似然估计法 &#x…
阅读更多...
Focal loss 损失函数详解

Focal loss 损失函数详解

Focal loss 目前目标检测的算法大致分为两类&#xff0c;One Stage 、Two Stage。 One Stage&#xff1a;主要指类似YOLO、SGD等这样不需要region proposal,直接回归的检测算法&#xff0c;这类算法检测速度很快&#xff0c;但是精度准确率不如使用Two stage的模型。 two St…
阅读更多...
机器学习之常见的损失函数(loss function)

机器学习之常见的损失函数(loss function)

解决一个机器学习问题主要有两部分&#xff1a;数据和算法。而算法又有三个部分组成&#xff1a;假设函数、损失函数、算法优化。我们一般在看算法书或者视频教学时&#xff0c;更多的是去推算或者说参数估计出其假设函数&#xff0c;而往往不太注重损失函数&#xff0c;但是损…
阅读更多...
深度学习loss函数理解

深度学习loss函数理解

机器学习中的范数规则化之L0、L1、L2范数及loss函数 监督机器学习问题无非就是“minimizeyour error while regularizing your parameters”&#xff0c;也就是在规则化参数的同时最小化误差。 最小化误差是为了让我们的模型拟合我们的训练数据&#xff0c;而规则化参数是防止我…
阅读更多...
CE Loss，BCE Loss以及Focal Loss的原理理解

CE Loss，BCE Loss以及Focal Loss的原理理解

一、交叉熵损失函数&#xff08;CE Loss&#xff0c;BCE Loss&#xff09; 最开始理解交叉熵损失函数被自己搞的晕头转向的&#xff0c;最后发现是对随机变量的理解有偏差&#xff0c;不知道有没有读者和我有着一样的困惑&#xff0c;所以在本文开始之前&#xff0c;先介绍一下…
阅读更多...
损失函数loss

损失函数loss

http://blog.csdn.net/pipisorry/article/details/23538535 监督学习及其目标函数 损失函数&#xff08;loss function&#xff09;是用来估量你模型的预测值f(x)与真实值Y的不一致程度&#xff0c;它是一个非负实值函数&#xff0c;通常使用L(Y, f(x))来表示。 损失函数是经…
阅读更多...
机器学习模型中的损失函数loss function

机器学习模型中的损失函数loss function

1. 概述 在机器学习算法中&#xff0c;有一个重要的概念就是损失函数&#xff08;Loss Function&#xff09;。损失函数的作用就是度量模型的预测值 f ( x ) f\left ( \mathbf{x} \right ) f(x)与真实值 y \mathbf{y} y之间的差异程度的函数&#xff0c;且是一个非负实值函数。…
阅读更多...
损失函数（Loss）

损失函数（Loss）

如果我们定义了一个机器学习模型&#xff0c;比如一个三层的神经网络&#xff0c;那么就需要使得这个模型能够尽可能拟合所提供的训练数据。但是我们如何评价模型对于数据的拟合是否足够呢&#xff1f;那就需要使用相应的指标来评价它的拟合程度&#xff0c;所使用到的函数就称…
阅读更多...
focal loss详解

focal loss详解

文章目录 focal loss的整体理解易分辨样本、难分辨样本的含义focal loss的出现过程focal loss 举例说明focal loss的 α \alpha α变体 focal loss的整体理解 focal loss 是一种处理样本分类不均衡的损失函数&#xff0c;它侧重的点是根据样本分辨的难易程度给样本对应的损失添…
阅读更多...
推荐文章

Copyright @ 2022~2023