异步编程-CompletableFuture

简介

支持异步计算函数的对象，提供异步，可返回，可编排的异步工具类。

是从 Java1.8 引入的新特性，位于 JUC^(java.util.concurrent)^ 包下中。

GPT 的解释是：

CompletableFuture 是 Java 8 引入的一个类，属于 java.util.concurrent 包，提供了一个强大的、异步编程的工具，用于表示未来某个时刻完成的计算结果。它的核心优势在于提供了一种简洁的方式来编写非阻塞的、基于回调的代码。

类继承关系图

技巧

结尾带：Async

一般以Async结尾的成员方法，指的是，开启新的线程运行

入参带：Executor

在指定的线程池上运行

结尾带：Accept

看见Accept结尾方法，代表无结果，有入参^（this本身）^

结尾带：Apply

看见Apply结尾的方法，代表有结果，有入参^（this本身）^

结尾带：Both

看见Both结尾的方法，代表等待两个Future完成，并且函数式接口需要处理两个结果^（this 和入参other）^，是否有结果取决于中段是Accept或者Apply

结尾带：Either

看见Either结尾的方法，代表竞争，this/other中有任意一个Future完成，则触发此方法

是否处理胜利者^（指竞争的胜利者，最先运行完毕的Future）^返回值取决于中段是Accept或者Apply

运行时机

当代码执行后，异步内容也会立即执行。

如果主线程即将运行完毕的情况，异步内容还是没有执行完毕，则主线程会直接结束，然后忽略异步内容运行的结果

方法列表

静态方法（public static）

静态方法数量: 12个^（基于JDK11 ）^

supplyAsync(Supplier supplier)
以异步的方式运行

特点：无入参，有返回值

相似方法：
1. supplyAsync(Supplier supplier,Executor executor)^(指定线程池运行)^

runAsync(Runnable runnable)

以异步的方式运行

特点：无入参，无返回值

相似方法：

runAsync(Runnable runnable,Executor executor)^(指定线程池运行)^

completedFuture(U value)

创建一个立即完成的CompletableFuture，它的返回值就是value参数

特点：立即创建一个立即完成的CompletableFuture

completedStage(U value)

和completedFuture(U value)差不多，但是它返回更通用的接口

特点：返回接口CompletionStage

allOf(CompletableFuture<?>... cfs)

创建一个新的CompletableFuture，它的完成状态是入参中的所有cf完成，则它完成。

如果cfs中出现异常，则它会直接返回异常，你可以用exceptionally来捕获异常，并执行相关操作

anyOf(CompletableFuture<?>... cfs)

创建一个新的CompletableFuture，它的完成状态是入参中任意一个cf完成，则它完成

它的返回值是完成CompletableFuture的返回值

当未完成时出现异常，则抛出该异常。

当已完成时出现异常，则忽略该异常。

特点：返回第一个完成的CompletableFuture的返回值

delayedExecutor(long delay, TimeUnit unit)

创建一个带延迟的线程池，如果不指定线程池，则使用它默认的线程池

特点：返回Executor，如果其他CompletableFuture以它运行，那么会先延迟指定时间后再执行

delayedExecut cutor)^(将带延迟的线程池，延迟后以指定的线程池运行)^

failedFuture(Throwable ex)

创建一个异常完成的CompletableFuture

特点：异常完成的CompletableFuture，可以用校验方法^（isDone(),isCancelled(),isCompletedExceptionally()）^来判断真实情况

failedStage(Throwable ex)^(返回更通用的接口)^

成员方法（public）

isDone()

是否已经完成

特点：无论正常完成还是异常完成都返回True
isCancelled()

是否已经取消

特点：被取消时返回Ture
isCompletedExceptionally()

是否异常完成

特点：抛出异常时返回True
get()

完成CompletableFuture并返回结果

特点：会抛出对应异常^（CancellationException - 如果此 future 被取消 ExecutionException - 如果此 future 异常完成 InterruptedException - 如果当前线程在等待时被中断）^
join()

完成CompletableFuture并返回结果

特点：会抛出对应异常^（CancellationException - 如果计算被取消 CompletionException - 如果这个 Future 异常完成或者完成计算抛出异常）^
get(long timeout, TimeUnit unit)

完成CompletableFuture，并等待返回结果，如果超时，会抛出异常

特点：等待返回结果，超时抛出异常^（CancellationException - 如果此未来被取消 ExecutionException - 如果此未来异常完成 InterruptedException - 如果当前线程在等待时被中断 TimeoutException - 如果等待超时）^
getNow(T valueIfAbsent)

立即获取结果，如果CompletableFuture未完成，则返回valueIfAbsent

注意：如果CompletableFuture是异常完成的，则会报错。

特点：无需等待立即拥有结果
toCompletableFuture()

可以将一个CompletionStage转换成CompletableFuture

特点：转换成CompletableFuture后，那么支持调用的方法会更多
exceptionally(Function<Throwable, ? extends T> fn)

异常捕获函数，fn支持一个参数并返回一个值。

特点：捕获上一个调用链上的异常，并给予新的(托底)返回值。
cancel(boolean mayInterruptIfRunning)

注意：此方法属于Future接口，一般用于FutureTask，在CompletableFuture中执行的，不会干扰已经运行的任务^（不会中断已经运行的任务）^。

取消CompletableFuture的执行，mayInterruptIfRunning代表如果已经执行，是否与该被中断
返回boolean类型，如果成功取消，则后续不能调用get() / join()，如果调用会抛出异常CancellationException
obtrudeValue(T value)

强制改变当前CompletableFuture的值，使它正常完成，就算它已经是完成的^（无论正常完成还是异常完成，调用后都变成'完成'）^

特点：不论状态，使其完成
obtrudeException(Throwable ex)

强制改变当前CompletableFuture的值，使它异常完成，就算它已经是完成的

特点：不论状态，使其异常

getNumberOfDependents()

获取当前CompletableFuture的被依赖数量，只有当前CompletableFuture未完成才能获得依赖计数

代码示例：

 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "1";
});
future.thenApply((a) -> a + "333");
future.thenApply((a) -> a + "444");
future.thenApply((a) -> a + "555");
System.out.println("future.getNumberOfDependents() = " + future.getNumberOfDependents());
// future.getNumberOfDependents() = 3

特点：获得当前被依赖的数量，方便管理

toString()

打印当前CompletableFuture状态，示例^（java.util.concurrent.CompletableFuture@1324409e[Not completed, 3 dependents]）^
newIncompleteFuture()

JDK1.9加入

创建一个新的CompletableFuture，底层实际上就是new了一个新的对象。
defaultExecutor()

返回默认的公共线程池
copy()

复制当前的CompletableFuture，并复制完成状态和返回值，新的CompletableFuture拥有独立的状态和返回值.

JDK1.9加入

特点：新的CompletableFuture拥有和this一样的状态和返回值，但独立与this
minimalCompletionStage()

将CompletableFuture转换为CompletionStage

貌似没有什么特点
orTimeout(long timeout, TimeUnit unit)

如果在指定的时间内没有完成，则以TimeoutException完成它。

特点：超时则异常完成
complete(T value)

用于给一个没有完成的CompletableFuture设置一个返回值，调用此方法后该CompletableFuture会设置为已完成状态，如果该CompletableFuture已经完成，则不会有任何变化^（可以使用obtrudeValue()来强行设置）^
1. completeAsync(Supplier<? extends T> supplier)^(一样，只是以异步方式运行)^
2. completeAsync(Supplier<? extends T> supplier, Executor executor)^(一样，不仅以异步的方式运行，还能指定线程池)^
3. completeExceptionally(Throwable ex)
  
  以异常的方式完成CompletableFuture
4. completeOnTimeout(T value, long timeout,TimeUnit unit)
  
  如果在规定的时间内，该CompletableFuture没有完成，则使用value完成它
thenApply(Function<? super T,? extends U> fn)

等待this完成，并返回一个新的CompletableFuture

特点：接受一个参数，返回一个结果，结果返回给新的CompletableFuture
1. thenApplyAsync(Function<? super T,? extends U> fn)^(以异步的方式完成)^
2. thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)^(以异步的方式完成，并指定线程池)^
thenAccept(Consumer<? super T> action)

等待this完成，并返回一个新的CompletableFuture，它是Void的

特点：接受一个参数，不返回结果，但返回一个新的CompletableFuture
1. thenAcceptAsync(Consumer<? super T> action)^(以异步的方式完成)^
2. thenAcceptAsync(Consumer<? super T> action, Executor executor)^(以异步的方式完成，并指定线程池)^
thenRun(Runnable action)

等待this完成，并返回一个新的CompletableFuture，它是Void的

特点：不接受参数，不返回结果，但返回一个新的CompletableFuture
1. thenRunAsync(Runnable action)^(以异步的方式完成)^
2. thenRunAsync(Runnable action, Executor executor)^(以异步的方式完成，并指定线程池)^
thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)

等待两个独立的^（this 和入参other）^CompletableFuture执行完毕后，合并两个^（this 和入参other）^CompletableFuture的结果，并返回新的带结果的CompletableFuture

特点：this是fn第一个参数，other是fn第二个参数，返回新的CompletableFuture
1. thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)^(以异步的方式完成)^
2. thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor)^(以异步的方式完成，并指定线程池)^
thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)

等待两个独立的^（this 和入参other）^CompletableFuture执行完毕后，处理两个CompletableFuture的结果，并返回新的不带结果的CompletableFuture

特点：this是fn第一个参数，other是fn第二个参数，返回新的不带结果的CompletableFuture
1. thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)^(以异步的方式完成)^
2. thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor)^(以异步的方式完成，并指定线程池)^
runAfterEither(CompletionStage<?> other,Runnable action)

等待两个独立的^（this 和入参other）^CompletionStage任意一个完成后运行后续处理，不处理它们的结果。

特点：竞争2选1，不关心结果
1. runAfterEitherAsync(CompletionStage<?> other,Runnable action)^(以异步的方式完成)^
2. runAfterEitherAsync(CompletionStage<?> other,Runnable action, Executor executor)^(以异步的方式完成，并指定线程池)^
runAfterBoth(CompletionStage<?> other,Runnable action)

等待两个独立的^（this 和入参other）^异步操作完成之后，做后续处理，但不关心它们的结果。

特点：全部完成，不关心结果
1. runAfterBothAsync(CompletionStage<?> other,Runnable action)^(以异步的方式完成)^
2. runAfterBothAsync(CompletionStage<?> other,Runnable action, Executor executor)^(以异步的方式完成，并指定线程池)^
applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)

等待两个独立的^（this 和入参other ）^CompletionStage任意一个完成后运行后续处理，处理竞争胜利的结果，并返回处理结果。

特点：竞争2选1，处理胜利者^（两个独立的异步操作优先完成的那个胜利者）^结果，并返回处理结果
1. applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn)^(以异步的方式完成)^
2. applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor)^(以异步的方式完成，并指定线程池)^
acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)

等待两个独立的^（this 和入参other ）^CompletionStage任意一个完成后运行后续处理，处理胜利者的结果，但本身不返回结果，它的返回值是CompletableFuture

特点：竞争2选1，处理胜利者^（两个独立的异步操作优先完成的那个胜利者）^结果，不返回结果
1. acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)^(以异步的方式完成)^
2. acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)^(以异步的方式完成，并指定线程池)^
thenCompose(Function<? super T, ? extends CompletionStage> fn)
等待this完成后，将this的结果作为入参，给fn使用，返回值是一个新的CompletableFuture，它和其他方法的区别在于，它的返回值必须是CompletableFuture
特点：链式，继续返回新的CompletableFuture

吐槽：感觉这玩意，一般用不上
1. thenComposeAsync(Function<? super T, ? extends CompletionStage> fn)^(以异步的方式完成)^

whenComplete(BiConsumer<? super T, ? super Throwable> action)

等待this完成，无论正常完成/异常完成，都会运行此方法，此方法的action接收两个参数。

特点：
action接受两个参数（result^(this的结果，异常的话此值为null)^,exception^(如果发生异常，则是异常信息，正常的话此值是null)^），如果在this在执行中异常，则会立即运行whenComplete方法，运行完成后再抛出异常，适用于异常后的清理。

用法：

    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            System.out.println("this运行了 = " + System.currentTimeMillis());
            int i = 1 / 0;
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "123";
        }).whenComplete((result, exception) -> {
            System.out.println("whenComplete运行了 = " + System.currentTimeMillis());
            System.out.println("result = " + result);
            System.out.println("exception = " + exception);
        }).join();

    }

	this运行了 = 1705582535741
	whenComplete运行了 = 1705582535751
	result = null
	exception = java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
	Exception in thread "main" java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
		at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)
		at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1702)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1692)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
Caused by: java.lang.ArithmeticException: / by zero
	at FutureTest.lambda$main$0(FutureTest.java:13)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
	... 6 more

whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action)^(以异步的方式完成)^
whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor)^(以异步的方式完成，并指定线程池)^

handle(BiFunction<? super T, Throwable, ? extends U> fn)

等待this完成，无论正常完成/异常完成，都会运行此方法，此方法的fn接收两个参数，此方法有返回值。

特点：action接受两个参数（result^(this的结果，异常的话此值为null)^,exception^(如果发生异常，则是异常信息，正常的话此值是null)^），返回新的结果和新的CompletableFuture

handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)^(以异步的方式完成)^
handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)^(以异步的方式完成，并指定线程池)^

异步编程-CompletableFuture

http://q-q.icu/archives/yi-bu-bian-cheng-completablefuture

作者

Administrator

发布于

2024年09月25日

许可协议

思源编辑器下一篇

异步编程-CompletableFuture

简介

类继承关系图

技巧

结尾带：Async

入参带：Executor

结尾带：Accept

结尾带：Apply

结尾带：Both

结尾带：Either

运行时机

方法列表

静态方法 （public static）

supplyAsync(Supplier supplier)

runAsync(Runnable runnable)

completedFuture(U value)

completedStage(U value)

allOf(CompletableFuture<?>... cfs)

anyOf(CompletableFuture<?>... cfs)

delayedExecutor(long delay, TimeUnit unit)

failedFuture(Throwable ex)

成员方法（public）

isDone()

isCancelled()

isCompletedExceptionally()

get()

join()

get(long timeout, TimeUnit unit)

getNow(T valueIfAbsent)

toCompletableFuture()

exceptionally(Function<Throwable, ? extends T> fn)

cancel(boolean mayInterruptIfRunning)

obtrudeValue(T value)

obtrudeException(Throwable ex)

getNumberOfDependents()

toString()

newIncompleteFuture()

defaultExecutor()

copy()

minimalCompletionStage()

orTimeout(long timeout, TimeUnit unit)

complete(T value)

completeExceptionally(Throwable ex)

completeOnTimeout(T value, long timeout,TimeUnit unit)

thenApply(Function<? super T,? extends U> fn)

thenAccept(Consumer<? super T> action)

thenRun(Runnable action)

thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)

thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)

runAfterEither(CompletionStage<?> other,Runnable action)

runAfterBoth(CompletionStage<?> other,Runnable action)

applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)

acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)

thenCompose(Function<? super T, ? extends CompletionStage> fn)

whenComplete(BiConsumer<? super T, ? super Throwable> action)

handle(BiFunction<? super T, Throwable, ? extends U> fn)

静态方法（public static）