当前位置：首页 > news >正文

JAVA：异步任务处理类CompletableFuture让性能提升一倍

news 来源：原创 2024/7/6 18:13:50

一、前言

CompletableFuture 是 Java 8 引入的一个功能强大的类，用于异步编程。它表示一个可能尚未完成的计算的结果，你可以对其添加回调函数来在计算完成时执行某些操作。在 Spring Boot 应用中，CompletableFuture 可以用于提高应用的响应性和吞吐量。

二、为什么使用异步编程

场景描述：进入用户个人中心，调用接口。接口需要返回用户基础信息、用户积分、用户等级等，假设用户基础信息查询耗时300ms，查询用户积分、用户等级分别耗时200ms，那么接口返回就需要700+ms了。那么我们如何去优化呢

如果采用异步（多线程并行）形式，接口将以用户基础信息查询耗时300ms最慢的为主，接口性能提升一倍！查询任务越多，则其性能提升越大！

三、异步编程介绍

在Java中，Callable、Runnable、Future和CompletableFuture是用于处理并发编程的重要接口和类。它们各自有不同的用途和特性，下面是对它们的简要介绍：

1. Callable
Callable是一个接口，用于定义返回结果或抛出异常的任务。它类似于Runnable，但Runnable没有返回值且不能抛出受检异常。Callable接口中定义了一个call()方法，该方法可以返回一个值，并且可以抛出一个受检异常。

2. Runnable
Runnable是一个接口，用于定义没有返回值且不抛出受检异常的任务。它只有一个run()方法，用于执行任务的代码。通常，Runnable对象会被传递给Thread对象的构造函数来创建新线程。

3. Future
Future是一个接口，用于表示异步计算的结果。它是Java并发包java.util.concurrent的一部分。当你提交一个Callable任务给ExecutorService时，它会返回一个Future对象，该对象表示异步计算的结果。你可以使用Future的get()方法来等待计算完成并获取结果，或者使用isDone()方法来检查计算是否完成。

4. CompletableFuture
CompletableFuture是Java 8中引入的一个类，实现了Future和CompletionStage接口。它提供了函数式编程的能力来处理异步编程，并允许你链式地组合多个异步操作。CompletableFuture提供了许多方法，如thenApply(), thenAccept(), thenCompose(), exceptionally()等，这些方法允许你定义当异步操作完成时应该执行的操作。

CompletableFuture相对于Future的优势在于它提供了更丰富的API来处理异步计算的结果，并支持链式调用和组合多个异步操作。这使得编写异步代码更加简洁和直观。

四、CompletableFuture中的函数式编程

在CompletableFuture类中的方法，很多都是Function函数式接口方法为入参，所以我们先要有一定的认识。

   Supplier<U>  // 生产者，没有入参，有返回结果Consumer<T>  // 消费者，有入参，但是没有返回结果Function<T,U>// 函数，有入参，又有返回结果

五、CompletableFuture核心方法介绍

5.1. 创建异步任务

supplyAsync: 异步执行一个给定的Supplier函数，并返回一个新的CompletableFuture，其结果由Supplier决定。

  CompletableFuture<Void/Type> supplyAsync(Supplier<U> supplier)CompletableFuture<Void/Type> supplyAsync(Supplier<U> supplier, Executor executor)：允许指定执行器。

runAsync: 异步执行一个Runnable任务，由于没有返回值，所以返回的CompletableFuture类型为Void。

  CompletableFuture<Void> runAsync(Runnable runnable)CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)：允许指定执行器。

5.1.1. 代码示例

supplyAsync

CompletableFuture<String>future=CompletableFuture.supplyAsync(()->{System.out.println("compute test");return "test";
});String result = future.join();System.out.println("get result: " + result);

runAsync

CompletableFuture<Void> future = CompletableFuture.runAsync(()->{System.out.println("compute test");
});System.out.println("get result: " + future.join());

指定线程池

ExecutorService executorService = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));CompletableFuture<User> task1 = CompletableFuture.supplyAsync(()->{User userInfo = userService.getUserInfo(1);return userInfo;
}, executorService);// 获取结果
User user = task1.get();

5.2. 流式连接函数介绍

流式连接函数（也称为组合函数）允许你以函数式编程的方式组合和链接多个异步操作。假如：我有两个任务，任务2需要等任务1执行完成后，拿到任务1中的结果作为任务2的参数，这时候就需要用到流式连接函数。

①. thenApply 和 thenApplyAsync

thenApply: 当前CompletableFuture计算完成时，应用给定的函数到其结果上，并返回表示该结果的新的CompletableFuture。

thenApplyAsync: 与thenApply类似，但异步执行给定的函数。

②. thenAccept 和 thenAcceptAsync

thenAccept: 当当前CompletableFuture计算完成时，执行给定的动作到其结果上，然后返回void。

thenAcceptAsync: 与thenAccept类似，但异步执行给定的动作。

③. thenRun 和 thenRunAsync

thenRun: 当当前CompletableFuture计算完成时，执行给定的动作，不关注其结果。

thenRunAsync: 与thenRun类似，但异步执行给定的动作。

④. thenCombine 和 thenCombineAsync

thenCombine: 接收另一个CompletableFuture，并返回一个新的CompletableFuture，该CompletableFuture在当前CompletableFuture和另一个CompletableFuture都完成计算后，使用它们的结果来计算值。

thenCombineAsync: 与thenCombine类似，但异步执行给定的函数。

⑤. thenCompose 和 thenComposeAsync

thenCompose: 当当前CompletableFuture计算完成时，应用给定的函数到其结果上，该函数返回一个新的CompletableFuture，并返回表示该新CompletableFuture的结果的CompletableFuture。

thenComposeAsync: 与thenCompose类似，但异步执行给定的函数。

⑥. whenComplete 和 whenCompleteAsync

whenComplete: 当当前CompletableFuture正常完成或出现异常时，执行给定的动作。动作有两个参数：结果（如果操作正常完成）或异常（如果抛出异常），以及一个表示该CompletableFuture的Throwable（如果操作抛出异常则为null）。

whenCompleteAsync: 与whenComplete类似，但异步执行给定的动作。

⑦. handle 和 handleAsync

handle: 当当前CompletableFuture正常完成或出现异常时，应用给定的函数到其结果或异常上，并返回一个新的CompletableFuture，其结果是函数的结果。

handleAsync: 与handle类似，但异步执行给定的函数。

注意：带Async后缀的函数表示需要连接的后置任务会被单独提交到线程池中，从而相对前置任务来说是异步运行的。除此之外，两者没有其他区别。

在使用Async后缀的方法时，你可以提供一个Executor作为可选参数来指定异步操作应在哪个线程上执行。如果未提供，则使用ForkJoinPool.commonPool()。

5.3. 流式连接核心函数代码示例

1. thenApply / thenAccept / thenRun

这组函数主要用于连接前后有依赖的任务链。这里将thenApply / thenAccept / thenRun放在一起讲，因为这几个连接函数之间的唯一区别是提交的任务类型不一样。

区别如下：

1.thenApply提交的任务类型需遵从Function签名，也就是有入参和返回值，其中入参为前置任务的结果。

2.thenAccept提交的任务类型需遵从Consumer签名，也就是有入参但是没有返回值，其中入参为前置任务的结果。

3.thenRun提交的任务类型需遵从Runnable签名，即没有入参也没有返回值。

代码示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{System.out.println("compute 1");return 1;
});CompletableFuture<Integer> future2 = future1.thenApply((p)->{System.out.println("compute 2");return p+10;
});
System.out.println("result: " + future2.join());

注意的是，通过thenApply / thenAccept / thenRun连接的任务，当且仅当前置任务计算完成时，才会开始后置任务的计算。

应用场景总结：

1. thenApply：当需要在异步操作的结果上执行额外的计算或转换时，使用 thenApply。

2. thenAccept：当只需要消费异步操作的结果，而不关心新的结果或执行额外的任务时，使用 thenAccept。

3. thenRun：当需要在异步操作完成后执行一个不依赖于其结果的任务时，使用 thenRun。

2. thenCombine

thenCombine最大的不同是连接任务可以是一个独立的CompletableFuture，从而允许前后连接的两个任务可以并行执行（后置任务不需要等待前置任务执行完成），最后当两个任务均完成时，再将其结果同时传递给下游处理任务，从而得到最终结果

代码示例：
假设我们有两个异步任务，一个用于获取用户的名字（nameFuture），另一个用于获取用户的年龄（ageFuture）。我们希望当这两个任务都完成后，将它们的结果组合成一个字符串，如 “Name: John, Age: 30”。


CompletableFuture<String> nameFuture = CompletableFuture.supplyAsync(() -> getUserName());  
CompletableFuture<Integer> ageFuture = CompletableFuture.supplyAsync(() -> getUserAge());  CompletableFuture<String> combinedFuture = nameFuture.thenCombine(ageFuture, (name, age) ->   "Name: " + name + ", Age: " + age  
);  combinedFuture.thenAccept(System.out::println); 
// 输出类似 "Name: John, Age: 30"

thenAcceptBoth、thenAcceptBothAsync、runAfterBoth、runAfterBothAsync的作用与thenConbime类似，区别如下：

1. thenAcceptBoth 和 thenAcceptBothAsync 用于处理两个异步操作的结果，但不返回新的结果。

2. runAfterBoth 和 runAfterBothAsync 也用于处理两个异步操作的完成，但不关注它们的结果。

3. thenCombine 用于处理两个异步操作的结果，并返回一个新的结果。

3. thenCompose

thenCompose 的应用场景主要涉及到需要基于一个异步任务的结果来发起另一个异步任务的情况。它允许你将多个异步操作链接在一起，并以前一个操作的结果作为后一个操作的输入。

代码示例：

如果使用thenApply实现如下：CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{System.out.println("compute 1");return 1;
});
CompletableFuture<CompletableFuture<Integer>> future2 =future1.thenApply((r)->CompletableFuture.supplyAsync(()->r+10));
System.out.println(future2.join().join());当连接的任务越多时，代码会变得越来越复杂，嵌套获取层级也越来越深。

使用thenCompose
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{System.out.println("compute 1");return 1;
});
CompletableFuture<Integer> future2 = future1.thenCompose((r)->CompletableFuture.supplyAsync(()->r+10));
System.out.println(future2.join());

4. whenComplete

whenComplete主要用于注入任务完成时的回调通知逻辑。这个解决了传统future在任务完成时，无法主动发起通知的问题。前置任务会将计算结果或者抛出的异常作为入参传递给回调通知函数。

代码示例：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {  // 模拟从数据库读取数据的异步操作  return "Data from database";  
});  future.whenComplete((result, exception) -> {  if (exception == null) {  // 操作成功，记录成功日志  System.out.println("Operation succeeded. Result: " + result);  } else {  // 操作失败，记录异常日志  System.err.println("Operation failed with exception: " + exception.getMessage());  }  // 在这里还可以执行其他与结果无关的操作，如清理资源、发送通知等  
});

5. handle

handle与whenComplete的作用有些类似，但是handle接收的处理函数有返回值，而且返回值会影响最终获取的计算结果。

代码示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{System.out.println("compute 1");return 1;
});CompletableFuture<Integer> future2 = future1.handle((r, e)->{if(e != null){System.out.println("compute failed!");return r;} else {System.out.println("received result is " + r);return r + 10;}
});System.out.println("result: " + future2.join());

六、CompletableFuture中获取异步任务结果介绍

异步任务执行完成之后，需要获取结果，有如下一些方法：

1. allOf 方法用于等待多个 CompletableFuture 任务全部完成。它接受一个 CompletableFuture 数组或列表作为参数，并返回一个新的 CompletableFuture。这个新的 CompletableFuture 会在所有给定的 CompletableFuture 都完成时完成，但它不包含任何原始任务的结果。

2. anyOf 方法与 allOf 类似，但它只等待给定的 CompletableFuture 中的一个完成。它同样接受一个 CompletableFuture 数组或列表作为参数，并返回一个新的 CompletableFuture。这个新的 CompletableFuture 会在任何一个给定的 CompletableFuture 完成时完成，但它不包含任何原始任务的结果。

3. join 方法用于等待 CompletableFuture 完成，并返回其结果（如果可用）。如果 CompletableFuture 尚未完成，则 join 会阻塞当前线程，直到它完成。如果 CompletableFuture 异常完成，则 join 会抛出与异常完成相对应的异常。

4. get 方法与 join 类似，也用于等待 CompletableFuture 完成并返回其结果。但是，get 方法可以接收一个可选的超时参数和一个时间单位，以便在指定的时间内等待 CompletableFuture 完成。如果 CompletableFuture 在指定的时间内没有完成，则 get 方法会抛出 TimeoutException。另外，如果 CompletableFuture 异常完成，则 get 方法会抛出 ExecutionException（包装了原始异常）或 InterruptedException（如果当前线程在等待时被中断）。

以上就是对CompletableFuture的一些详细介绍，以及一些常用api的代码示例，希望对你在项目进行性能方面的代码优化中有一定的作用。