CompletableFuture使用详解

作者: 晚风吹___ | 来源:发表于2022-04-21 17:07 被阅读0次

什么是 CompletableFuture？

CompletableFuture 用于 Java 中的异步编程。异步编程是一种编写非阻塞代码的方法，方法是在与主应用程序线程不同的线程上运行任务，并通知主线程其进度、完成或失败。
这样，您的主线程不会阻塞/等待任务的完成，它可以并行执行其他任务。拥有这种并行性极大地提高了程序的性能。

我们首先看看CompletableFuture的类图关系，CompletableFuture实现了Future和CompletionStage接口，因此看来CompletableFuture具有Future和CompletionStage的特性

Dingtalk_20220421142622.jpg

Future vs CompletableFuture

CompletableFuture 是Java 5 中引入的 Java Future API的扩展。

Future 用作对异步计算结果的引用。它提供了isDone()一种检查计算是否完成的get()方法，以及一种在计算完成时检索计算结果的方法。

Future API 是向 Java 异步编程迈出的一大步，但它缺乏一些重要且有用的特性

Future局限性

无法手动完成：

假设您编写了一个接口来从远程 API 获取商品信息。由于此 API 调用非常耗时，因此您在单独的线程中运行它并从您的函数中返回一个 Future。

现在，让我们假设，如果远程API服务关闭了，那么您希望根据商品的最后一次缓存价格手动完成Future。你能在Future上做到这一点吗?不!
你不能在没有阻塞的情况下对 Future 的结果执行进一步的操作：

Future 不会通知你它的完成。它提供了一个阻塞get()方法，直到结果可用。

您无法将回调函数附加到 Future 并在 Future 的结果可用时自动调用它。
多个Future不能链接在一起：

有时您需要执行长时间运行的计算，当计算完成后，您需要将其结果发送到另一个长时间运行的计算，依此类推。
您不能将多个Future组合在一起：

假设您有 10 个不同的 Futures 想要并行运行，然后在它们全部完成后运行某个函数。你不能用 Future 做到这一点。
无异常处理：

Future API 没有任何异常处理结构。

CompletableFuture提供的静态方法

// 无返回值
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)

supply开头：这种方法，可以返回异步线程执行之后的结果。
run开头：这种不会返回结果，就只是执行线程任务。
如果你想异步运行一些后台任务并且不想从任务中返回任何东西，那么你可以使用run开头的

举个例子：

CompletableFuture.runAsync(() -> System.out.println("执行无返回结果的异步任务"));
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("执行有返回值的异步任务");
    return "hello";
});

我们可以通过get()或者join()方法来获取返回的结果

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("执行有返回值的异步任务");
    return "hello";
});
future.get();

输出：hello
那么这里我们总结一下

join&get&getNow的区别
join()与get() 区别在于join() 返回计算的结果或者抛出一个unchecked异常(CompletionException)，
而get() 返回一个具体的异常(ExecutionException, InterruptedException).
getNow() 则有所区别，也就是返回当前执行好的结果如果当前为执行完则
参数valueIfAbsent的意思是当计算结果不存在或者当前时刻没有完成任务，给定一个确定的值。

getNow()举个例子：

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
// 模拟业务时长
ThreadUtil.sleep(500000);
System.out.println("执行有返回值的异步任务");
return "你好 2022";   
});
System.out.println(future2.getNow("没有完成任务"));

输出：没有完成任务

但是注意！！!该CompletableFuture.get()方法是阻塞的。它一直等到 Future 完成并在完成后返回结果。但是，这样不相当同步阻塞了嘛？这不是我们想要的。我们想真正构建我们的异步操作。

可以使用thenApply(),thenAccept(),thenRun()方法附加到CompletableFuture上

1. thenApply()方法

可以使用thenApply()方法在 CompletableFuture 到达时对其进行处理和转换。它将Function < T,R >作为参数。Function < T,R >是一个简单的函数式接口，表示一个接受 T 类型参数并产生 R 类型结果的函数

什么意思呢，也就是拿到上一个异步线程返回的结构进行后续处理
举个例子

打个比方我想获取订单的详情页，那么订单详情中间又有一个获取商品信息的操作，那么我需要一直等待等到拿到商品信息之后才进行订单详情的操作嘛，这样不就是同步并阻塞的吗。那么我们可以获商品信息之后再thenApply()

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        // 模拟业务时长   
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {          
        throw new IllegalStateException(e);     
    }
    return "商品信息";
});

    CompletableFuture<String> resultFuture = future.thenApply(name -> "订单信息 " + name);
    System.out.println(resultFuture.get());

输出：订单信息商品信息订单编号信息

我们总结一下：

thenApply()是线程的后续操作，并且有这么一个特性，可以拿到上一次线程执行的返回结果这本次thenApply()的参数一直传递下去。并且是有返回结果的。

小提示：方法中带Async后缀的都是异步操作例如：thenApplyAsync()，以下同理。

2. thenAccept() 和 thenRun()方法

如果你不想从你的回调函数中返回任何东西，只想在 Future 完成后运行一些代码，那么你可以使用thenAccept()andthenRun()方法。这些方法是消费者Consumer<? super T> action，通常用作回调链中的最后一个回调。

举个例子：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "商品信息");
future.thenAccept(product  -> System.out.println("我拿到了 " + product));

输出：我拿到了商品信息

但是thenRun()无法拿到上一次的结果，thenRun()意为然后执行什么，里面接收一个Runnable参数

最后贴出这些接口：

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)

public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

关于方法里面的参数Executor问题

runAsync方法和supplyAsync方法第二个参数需要指定线程池Executor，如果不指定，会使用默认的线程池ForkJoinPool，关于什么是ForkJoinPool，可以自行查阅，大致是：

ForkJoinPool就是用来解决这种问题的：将一个大任务拆分成多个小任务后，
使用fork可以将小任务分发给其他线程同时处理，使用join可以将多个线程处理的结果进行汇总；这实际上就是分治思想的并行版本。

这些线程都是Daemon线程，主线程结束Daemon线程不结束，只有JVM关闭时，生命周期终止

当然我们也可以自己实现线程池

3.complete()

注意complete()其实也是个消费操作，但是与thenRun()不同的是，里面可以可抛出的异常

// 区别就是不是异步处理
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)

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            int a = 10 / 0;
            System.out.println("执行结束！");
            return "hello";
        });

        future.whenComplete((t, action) -> System.out.println(t + " 执行完成！"));

        // 异常时执行
        future.exceptionally(t -> {
            System.out.println("执行失败：" + t.getMessage());
            return "异常";
        }).join();

执行结果：

null 执行完成！
执行失败：java.lang.ArithmeticException: / by zero

Process finished with exit code 0

其实说白了就是拿到上个阶段线程执行的结果当作下个阶段续操作的参数处理

4.handle()

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)

handle方法集和上面的complete方法集没有区别，同样有两个参数一个返回结果和可抛出异常，区别就在于返回值

CompletableFuture 组合

假如你想做一个详情页，里面有用户信息，商品信息，你想把他们组合到一起

当然我们可以使用上面说到的thenApply()操作，也可以使用thenCompose()

举个例子：

使用thenApply()

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "商品信息");
CompletableFuture<String> total = future.thenApply(product -> product + " 用户信息");
System.out.println(total.join());

输出：商品信息用户信息

使用thenCompose()

CompletableFuture<String> productFuture = CompletableFuture.supplyAsync(() -> {
    // 第一个任务拿到商品信息
    return "商品信息";
}).thenCompose(product -> {
    // 向第二个任务传递参数list(上一个任务的信息)
    return CompletableFuture.supplyAsync(() -> product + " 用户信息");
});

System.out.println(productFuture.get());

输出：商品信息用户信息

有什么区别呢？

上述情况的最终结果是一个嵌套的 CompletableFuture，什么是嵌套?也就是total是最终返回结果total嵌套着future
如果您希望最终结果是顶级 Future,用thenCompose()方法

多个 CompletableFuture 组合在一起

假设有3个接口获取用户信息，获取商品信息，获取会员信息,先每个接口1s,那个3个接口总耗时3s。按顺序执行此操作，但这将花费大量时间，

所以异步线程会大大减少耗时，增加程序的性能

CompletableFuture.allOf()

CompletableFuture.allOf()里面包含了多个CompletableFuture操作，查阅源码

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
    return andTree(cfs, 0, cfs.length - 1);
}

接收一个CompletableFuture类型的可变参数。

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "用户信息");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "商品信息");
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "会员信息");

CompletableFuture.allOf(future1, future2, future3);
System.out.println(future1.join());
System.out.println(future2.join());
System.out.println(future3.join());

输出：

用户信息
商品信息
会员信息
我们可以再看

 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
     // 模拟业务时长
     ThreadUtil.sleep(2000);
     System.out.println("future1当前线程：" + Thread.currentThread().getId());
     
     return "用户信息";
 });
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    System.out.println("future2当前线程：" + Thread.currentThread().getId());
    return "商品信息";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    System.out.println("future3当前线程：" + Thread.currentThread().getId());
    return "会员信息";
});

假设用户接口中间做了一些业务操作，消耗了时长，那么想想，不可能总是等吧，当然不是，这里CompletableFuture会开启多个线程去执行操作。我们打印线程。

输出

future2当前线程：12
future3当前线程：14
future1当前线程：13
用户信息
商品信息
会员信息

可以看到线程id是不一样的

CompletableFuture.anyOf()

顾名思义是任何的意思，也就是多个任务中哪个任务先返回我就返回结果。

有的时候可能每个接口执行的时长是不同的，那我们可能需要耗时最短的结果作为代替时

举个例子：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    // 模拟业务时长
    ThreadUtil.sleep(3000);
    return "用户信息";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    // 模拟业务时长
    ThreadUtil.sleep(2000);
    return "商品信息";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    return "会员信息";
});

CompletableFuture<Object> future = CompletableFuture.anyOf(future1, future2, future3);
System.out.println(future.join());

输出结果：会员信息

CompletableFuture 异常处理

如果在执行过程中害怕出错，可以加上异常处理

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    int age = -1;
    if(age < 0) {
        throw new IllegalArgumentException("Age can not be negative");
    }
    if(age > 18) {
        return "成年人";
    } else {
        return "小孩";
    }
}).exceptionally(ex -> {
    System.out.println("年龄错误 " + ex.getMessage());
    return "Unknown!";
});
System.out.println(future.join());

输出：

年龄错误 java.lang.IllegalArgumentException: Age can not be negative
Unknown!

当然handle()方法也可以处理异常

 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
     int age = -1;
     if(age < 0) {
         throw new IllegalArgumentException("Age can not be negative");
     }
     if(age > 18) {
         return "成年人";
     } else {
         return "小孩";
     }
 }).handle((res, ex) -> {
     System.out.println("年龄错误: " + ex.getMessage());
     System.out.println("res: " + res);
     return "Unknown!";
 });
System.out.println(future.join());

输出：

年龄错误: java.lang.IllegalArgumentException: Age can not be negative
res: null
Unknown!

如果发生异常，则res参数将为 null，否则，ex参数将为 null。
以上还有很多不足之处，欢迎补充

CompletableFuture使用详解

什么是 CompletableFuture？

Future vs CompletableFuture

Future局限性

CompletableFuture提供的静态方法

1. thenApply()方法

2. thenAccept() 和 thenRun()方法

3.complete()

4.handle()

CompletableFuture 组合

多个 CompletableFuture 组合在一起

CompletableFuture.allOf()

CompletableFuture.anyOf()

CompletableFuture 异常处理

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

收藏

一些收藏

java基础