Java | CompletableFuture详解

news/2025/2/4 7:55:31/

关注：CodingTechWork

CompletableFuture 概述

介绍

  CompletableFuture是 Java 8 引入的一个非常强大的类，属于 java.util.concurrent 包。它是用于异步编程的一个工具，可以帮助我们更方便地处理并发任务。与传统的线程池或 Future 对比，CompletableFuture 提供了更多灵活性和组合功能，使得异步编程更加简单和易于维护。
  CompletableFuture主要用于异步操作和组合多个异步任务。它可以通过执行非阻塞的操作来避免阻塞主线程，从而提高程序的性能和响应速度。
  CompletableFuture实现了 Future 和 CompletionStage 接口。

优势

相比传统的 Future 接口，具有以下核心优势：

  • 支持非阻塞的异步操作
  • 提供链式调用和组合操作的能力
  • 内置完善的异常处理机制
  • 支持函数式编程风格

与传统 Future 对比

特性FutureCompletableFuture
结果获取阻塞 get() 方法支持回调通知
链式调用不支持支持多级流水线处理
异常处理需要 try-catch内置异常传播机制
组合操作手动实现提供多种组合方法
手动完成不支持支持 complete() 方法

核心方法

supplyAsync()

用于执行一个异步任务并返回一个结果。它接受一个 Supplier，并在后台线程执行该任务。

java">CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {return 42;
});

runAsync()

用于执行一个异步任务，但不返回结果。它接受一个Runnable，并在后台线程执行该任务。

java">CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {System.out.println("Task executed");
});

thenApply()

用于将前一个计算的结果转换为另一个结果。它接受一个Function，该函数作用于先前计算的结果，并返回一个新的CompletableFuture。

java">CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 5).thenApply(result -> result * 2);

thenAccept()

用于处理前一个计算结果并返回Void。它接受一个Consumer，用于对先前的结果执行操作。

java">CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> 5).thenAccept(result -> System.out.println(result));

thenCombine()

用于将两个CompletableFuture的结果结合成一个新的结果。它接受两个CompletableFuture和一个BiFunction。

java">CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 5);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 10);CompletableFuture<Integer> combined = future1.thenCombine(future2, (result1, result2) -> result1 + result2);

thenCompose()

用于将两个异步操作串联起来，第二个任务依赖第一个任务的结果。它接受一个Function，返回一个新的CompletableFuture。

java">CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 5).thenCompose(result -> CompletableFuture.supplyAsync(() -> result * 2));

allOf()

用于组合多个CompletableFuture，并等待它们全部完成。返回一个新的 CompletableFuture，它完成时表示所有的CompletableFuture都已完成。

java">CompletableFuture<Void> future1 = CompletableFuture.supplyAsync(() -> 5);
CompletableFuture<Void> future2 = CompletableFuture.supplyAsync(() -> 10);CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2);

anyOf()

用于组合多个CompletableFuture，并等待任意一个任务完成。返回一个新的 CompletableFuture，该CompletableFuture完成时表示至少有一个任务已完成。

java">CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 5);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 10);CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2);

exceptionally()

用于在异步任务执行过程中发生异常时处理异常。它接受一个Function，如果任务执行失败（抛出异常），则返回一个备用值。

java">CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {if (true) throw new RuntimeException("Error");return 5;})// 处理异常，返回-1.exceptionally(ex -> -1);

whenComplete()

用于在任务完成后进行额外操作，不论任务是正常完成还是异常完成。它接受一个 BiConsumer，用于处理结果和异常。

java">CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 5).whenComplete((result, ex) -> {if (ex == null) {System.out.println("Completed with result: " + result);} else {System.out.println("Completed with exception: " + ex.getMessage());}});

supplyAsync()和runAsync() 区别

CompletableFuture.supplyAsync() 和 CompletableFuture.runAsync() 都是用来异步执行任务的，但它们之间有一些关键区别，主要体现在是否返回结果以及如何处理任务：

返回值

  • supplyAsync()功能：该方法接受一个 Supplier 函数，Supplier 是一个能够返回结果的函数。它会异步执行任务并返回一个 CompletableFuture<T>，这个 CompletableFuture 可以通过 get() 等方法获取任务的返回值。
  • supplyAsync()适用场景：当你需要执行一个任务，并且希望获取这个任务的结果时，使用 supplyAsync()。
  • runAsync()功能：该方法接受一个 Runnable函数，Runnable是一个没有返回值的任务。它会异步执行任务，但不会返回任何结果，因此返回的是一个CompletableFuture<Void>，意味着你只能知道任务是否完成，但无法直接获取任务的结果。
  • runAsync()适用场景：当你只需要执行任务，但不关心返回结果时，使用runAsync()。

任务执行

  • supplyAsync()：异步任务执行后会返回一个值，通常使用 thenApply()、thenAccept() 等方法来处理返回值。
  • runAsync()：执行的是没有返回值的任务，通常用 thenRun() 来处理任务完成后的操作。

使用场景

  • supplyAsync()：适用于有返回值的任务，例如你需要异步计算某个结果。
  • runAsync()：适用于没有返回值的任务，例如日志记录、状态更新等。

技术原理

创建异步任务

java">// 使用默认 ForkJoinPool
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "Result");// 指定自定义线程池
ExecutorService executor = Executors.newFixedThreadPool(2);
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {// 业务逻辑return "Custom Result";
}, executor);

结果处理

同步回调

java">// thenApply: 同步转换结果
cf.thenApply(s -> s + " processed").thenAccept(System.out::println);// thenAccept: 同步消费结果
cf.thenAccept(result -> {System.out.println("Received: " + result);
});

异步回调

java">// thenApplyAsync: 异步转换
cf.thenApplyAsync(s -> {// 在 ForkJoinPool 异步执行return s.toUpperCase();
});// 指定自定义线程池
cf.thenAcceptAsync(result -> {System.out.println("Async processing");
}, executor);

组合操作

串行组合

java">// 查询用户信息，然后根据用户ID查询订单信息（串行执行）
CompletableFuture<String> cf = queryUserInfo()// 使用 thenCompose 实现串行组合.thenCompose(user -> queryOrder(user.getId()));

并行组合

java">// 并行查询服务A和服务B
// 查询服务A
CompletableFuture<String> cf1 = queryServiceA(); 
// 查询服务B
CompletableFuture<String> cf2 = queryServiceB(); // 合并两个结果（并行执行，结果合并）
CompletableFuture<String> combined = cf1// 使用 thenCombine 合并两个任务的结果.thenCombine(cf2, (res1, res2) -> res1 + " & " + res2); // 任意一个完成即返回（并行执行，取最先完成的结果）
// 使用 anyOf 实现快速返回
CompletableFuture<Object> anyOf = CompletableFuture.anyOf(cf1, cf2);

异常处理

java">// 异常处理：捕获异常并返回默认值
cf.exceptionally(ex -> {// 打印异常信息System.err.println("Error: " + ex.getMessage()); // 返回默认值return "default value"; 
});// 异常处理：捕获异常并返回 fallback 值
cf.handle((result, ex) -> {// 如果有异常if (ex != null) { // 返回 fallback 值return "fallback"; }// 否则返回正常结果return result; 
});

超时控制

java">// 超时控制：如果任务在指定时间内未完成，返回默认值
// 1秒后未完成则返回 "default"
cf.completeOnTimeout("default", 1, TimeUnit.SECONDS);

典型应用场景

并行任务处理

java">// 并行处理多个请求
List<CompletableFuture<String>> futures = requests.stream().map(request -> // 对每个请求异步处理CompletableFuture.supplyAsync(() -> process(request), executor)) // 收集所有任务.collect(Collectors.toList()); // 等待所有任务完成
CompletableFuture<Void> allDone = CompletableFuture// 使用 allOf 等待所有任务完成.allOf(futures.toArray(new CompletableFuture[0]));

服务调用编排

java">// 服务调用编排：依次执行多个异步任务
CompletableFuture<Order> orderFuture = // 获取用户信息getUserProfile() // 验证地址.thenCompose(user -> validateAddress(user)) // 创建订单.thenCompose(address -> createOrder(address)) // 异常时创建备用订单.exceptionally(ex -> createFallbackOrder());

异步结果聚合

java">// 异步结果聚合：合并两个异步任务的结果
CompletableFuture<Integer> total = // 任务1：返回10CompletableFuture.supplyAsync(() -> 10) // 任务2：返回20，合并结果为30.thenCombine(CompletableFuture.supplyAsync(() -> 20), Integer::sum);

demo：电商订单处理

场景需求

  1. 并行查询用户信息和商品库存
  2. 验证地址有效性
  3. 组合结果创建订单
  4. 记录操作日志

实现代码

java">import java.util.concurrent.*;public class OrderService {private final ExecutorService executorService;// 构造函数接受线程池大小参数public OrderService(int poolSize) {// 创建一个自定义线程池，大小为 poolSizethis.executorService = Executors.newFixedThreadPool(poolSize);System.out.println("Thread pool initialized with size: " + poolSize);}// 线程池大小配置方法public void setThreadPoolSize(int poolSize) {// 关闭现有线程池并创建一个新的线程池shutdown();  // 关闭旧线程池this.executorService = Executors.newFixedThreadPool(poolSize);  // 创建新的线程池System.out.println("Thread pool resized to: " + poolSize);}// 获取当前线程池大小public int getThreadPoolSize() {if (executorService instanceof ThreadPoolExecutor) {return ((ThreadPoolExecutor) executorService).getCorePoolSize();}return -1; // 如果没有获取到线程池大小，则返回 -1}public CompletableFuture<OrderResult> createOrderAsync(OrderRequest request) {// 并行查询用户信息和库存信息CompletableFuture<UserInfo> userFuture = queryUserAsync(request.getUserId());CompletableFuture<Inventory> inventoryFuture = queryInventoryAsync(request.getSkuId());// 使用 thenCombine 合并两个异步结果return userFuture.thenCombine(inventoryFuture, (user, inventory) -> {// 验证用户地址validateAddress(user.getAddress());// 检查库存是否足够checkInventory(inventory);// 根据用户信息、库存信息和订单请求生成订单return generateOrder(user, inventory, request);})// 使用 thenCompose 对生成的订单进行进一步处理.thenCompose(order -> {// 异步保存订单return saveOrderAsync(order)// 保存成功后，发送通知.thenApply(savedOrder -> sendNotification(savedOrder));})// 处理最终结果，或者处理异常.handle((result, ex) -> {// 如果出现异常，记录错误并返回失败的结果if (ex != null) {log.error("Order failed", ex);return OrderResult.failure(ex.getMessage());}// 如果成功，返回成功的结果return OrderResult.success(result);});}// 异步查询用户信息，使用自定义线程池private CompletableFuture<UserInfo> queryUserAsync(String userId) {return CompletableFuture.supplyAsync(() -> userService.getUser(userId), executorService);}// 异步查询库存信息，使用自定义线程池private CompletableFuture<Inventory> queryInventoryAsync(String skuId) {return CompletableFuture.supplyAsync(() -> inventoryService.getStock(skuId), executorService);}// 验证用户地址是否合法private void validateAddress(String address) {if (address == null || address.isEmpty()) {throw new IllegalArgumentException("Address is invalid.");}// 进一步的地址验证逻辑}// 检查库存是否足够private void checkInventory(Inventory inventory) {if (inventory == null || inventory.getStock() <= 0) {throw new IllegalArgumentException("Insufficient inventory.");}}// 根据用户和库存信息生成订单private Order generateOrder(UserInfo user, Inventory inventory, OrderRequest request) {Order order = new Order();order.setUserId(user.getUserId());order.setSkuId(request.getSkuId());order.setQuantity(request.getQuantity());order.setTotalPrice(inventory.getPrice() * request.getQuantity());order.setShippingAddress(user.getAddress());order.setStatus(OrderStatus.PENDING);return order;}// 异步保存订单，使用自定义线程池private CompletableFuture<Order> saveOrderAsync(Order order) {return CompletableFuture.supplyAsync(() -> orderService.saveOrder(order), executorService);}// 发送订单通知private String sendNotification(Order order) {notificationService.sendOrderConfirmation(order);return "Order placed successfully!";}// 关闭线程池public void shutdown() {if (executorService != null && !executorService.isShutdown()) {executorService.shutdown();System.out.println("Thread pool shut down gracefully.");}}// 强制关闭线程池public void shutdownNow() {if (executorService != null && !executorService.isShutdown()) {executorService.shutdownNow();System.out.println("Thread pool shut down immediately.");}}// 线程池是否已关闭public boolean isShutdown() {return executorService.isShutdown();}// 主方法示例public static void main(String[] args) {// 初始化 OrderService 并传入线程池大小OrderService orderService = new OrderService(5);// 获取并输出当前线程池大小int poolSize = orderService.getThreadPoolSize();System.out.println("Current thread pool size: " + poolSize);// 假设创建一个订单请求对象OrderRequest orderRequest = new OrderRequest("user123", "sku456", 2);orderService.createOrderAsync(orderRequest).thenAccept(result -> {System.out.println(result.getMessage());});// 动态调整线程池大小orderService.setThreadPoolSize(10);System.out.println("Updated thread pool size: " + orderService.getThreadPoolSize());// 关闭线程池orderService.shutdown();}
}

输出结果

java">Thread pool initialized with size: 5
Current thread pool size: 5
Order placed successfully!
Thread pool resized to: 10
Updated thread pool size: 10
Thread pool shut down gracefully.

最佳实践与注意事项

线程池选择策略

  • CPU密集型任务使用有界线程池
  • IO密集型任务使用缓存线程池
  • 避免混合使用不同任务类型
    资源管理
java">try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {CompletableFuture.runAsync(() -> task(), executor);
}

调试技巧

  • 使用 thenApplyAsync 添加日志点
  • 包装异步操作添加跟踪ID
java">cf.thenApplyAsync(result -> {log.debug("[Trace-{}] Step completed", traceId);return result;
});

性能优化

  • 避免过度嵌套回调
  • 及时关闭自定义线程池
  • 使用CompletableFuture#join()谨慎

总结

CompletableFuture为 Java 异步编程提供了强大支持，特别适用于：

  • 需要编排多个异步操作的场景
  • 实现非阻塞的响应式系统
  • 需要精细控制任务执行顺序和依赖关系
  • 构建高并发、低延迟的服务

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

相关文章

C#魔法秘籍：委托与事件，开启多态回调与消息派对之旅

C#魔法秘籍：委托与事件，开启多态回调与消息派对之旅

一、引言&#xff1a;踏入 C# 魔法世界 嘿&#xff0c;各位编程小伙伴们&#xff01;欢迎来到 C# 的奇幻世界&#xff0c;今天我们要一起探索两个超级有趣又强大的特性 —— 委托&#xff08;Delegate&#xff09;和事件&#xff08;Event&#xff09;。这两个家伙可是 C# 编程…
阅读更多...
跨平台文件互传工具

跨平台文件互传工具

一款高效便捷的文件互传工具&#xff0c;支持在线快速传输各种文件格式&#xff0c;无需注册&#xff0c;直接分享文件。适用于个人和团队间的文件共享&#xff0c;跨平台支持&#xff0c;轻松解决文件传输问题。免费的文件传输服务&#xff0c;让你的工作更高效。 gotool
阅读更多...
redis原理之数据结构

redis原理之数据结构

dict dict&#xff0c;哈希表&#xff0c;redis 所有的 key-value 都存储在里面。如果曾经学过哈希表这种数据结构&#xff0c;那么很容易能写出一个来&#xff0c;但 redis dict 考虑了更多的功能。 // 哈希表&#xff08;字典&#xff09;数据结构&#xff0c;redis 的所有键…
阅读更多...
3、从langchain到rag

3、从langchain到rag

文章目录 本文介绍向量和向量数据库向量向量数据库 索引开始动手实现rag加载文档数据并建立索引将向量存放到向量数据库中检索生成构成一条链 本文介绍 从本节开始&#xff0c;有了上一节的langchain基础学习&#xff0c;接下来使用langchain实现一个rag应用&#xff0c;并稍微…
阅读更多...
C++：多继承习题3

C++：多继承习题3

题目内容&#xff1a; 声明一个时间类Time&#xff0c;时间类中有3个私有数据成员(Hour&#xff0c;Minute&#xff0c;Second)和两个公有成员函数(SetTime和PrintTime)。要求&#xff1a; &#xff08;1&#xff09; SetTime根据传递的3个参数为对象设置时间&#xff1b; &a…
阅读更多...
基于springboot+vue的航空散货调度系统

基于springboot+vue的航空散货调度系统

开发语言&#xff1a;Java框架&#xff1a;springbootJDK版本&#xff1a;JDK1.8服务器&#xff1a;tomcat7数据库&#xff1a;mysql 5.7&#xff08;一定要5.7版本&#xff09;数据库工具&#xff1a;Navicat11开发软件&#xff1a;eclipse/myeclipse/ideaMaven包&#xff1a;…
阅读更多...
【机器学习】深入无监督学习分裂型层次聚类的原理、算法结构与数学基础全方位解读，深度揭示其如何在数据空间中构建层次化聚类结构

【机器学习】深入无监督学习分裂型层次聚类的原理、算法结构与数学基础全方位解读，深度揭示其如何在数据空间中构建层次化聚类结构

&#x1f31f;个人主页&#xff1a;落叶 &#x1f31f;当前专栏: 机器学习专栏 目录 引言 分裂型层次聚类&#xff08;Divisive Hierarchical Clustering&#xff09; 1. 基本原理 2. 分裂型层次聚类的算法步骤 Step 1: 初始化 Step 2: 选择分裂的簇 Step 3: 执行分裂操作…
阅读更多...
CSS 中调整元素大小的全面指南

CSS 中调整元素大小的全面指南

CSS 中调整元素大小的全面指南 1. 原始尺寸&#xff08;固有尺寸&#xff09;示例代码&#xff1a;图像的固有尺寸 2. 设置具体的尺寸示例代码&#xff1a;设置固定宽度和高度 3. 使用百分比示例代码&#xff1a;使用百分比设置宽度 4. 使用百分比作为外边距和内边距示例代码&a…
阅读更多...
最新文章

Copyright @ 2022~2023