深入理解 CompletableFuture：Java 多线程编排与异步编程实践

1. 引言

在现代软件开发中，性能和响应速度往往决定了用户体验的好坏，而这两者又与高效的多线程处理密不可分。在 Java 中，CompletableFuture 是 JDK 8 引入的一种强大的工具，它不仅支持异步任务的执行，还能通过丰富的 API 实现复杂的多线程编排。

1.1 什么是 CompletableFuture

CompletableFuture 是 Java 标准库中提供的一个实现了 Future 接口的类，它的目标是让异步编程更加简单和灵活。与传统的 Future 不同，CompletableFuture 支持：

  • 非阻塞式操作：不需要显式调用 get 等阻塞方法等待结果。
  • 多步骤任务：支持任务的链式调用和依赖关系管理。
  • 异常处理：提供了全面的异常捕获和恢复机制。
  • 并行处理：允许多个异步任务并行执行并合并结果。
1.2 应用场景

CompletableFuture 的典型应用场景包括但不限于：

  • 微服务调用：在分布式系统中并发调用多个微服务，并将结果整合返回。
  • 批量处理：对大批量的数据进行并行计算和汇总。
  • 高性能 Web 应用：优化请求的响应时间，提升吞吐量。
  • 事件驱动架构：实现事件的异步处理和响应。
1.3 多线程编排的重要性

在开发过程中，简单的异步任务通常可以通过单独的线程池完成，但当任务之间存在复杂的依赖关系时，传统方式容易导致以下问题：

  • 代码复杂度提升：过多的嵌套和回调导致代码可读性下降。
  • 资源浪费：未合理控制并发，可能导致线程池耗尽或性能瓶颈。
  • 错误难以追踪：异常处理分散，导致问题难以排查。

而 CompletableFuture 提供了一种优雅的方式来解决这些问题，它支持以声明式的风格编写任务逻辑，显著提升代码的可读性和维护性。

2. 基础知识

在深入探讨 CompletableFuture 的多线程编排实践之前，我们需要对其基础知识有一定的了解。理解 CompletableFuture 的基本操作和底层概念，是高效使用它的前提。

2.1 Java 异步编程概述

在传统 Java 编程中，我们通常使用线程池 (ExecutorService) 来管理并发任务。然而，这种方法存在一定的局限性，例如：

  • 需要手动管理任务依赖和结果处理。
  • 异步任务完成后，结果的获取通常通过阻塞式的 get() 方法。
  • 异常处理复杂且容易遗漏。

CompletableFuture 通过链式 API 和函数式编程范式解决了上述问题，它支持以非阻塞的方式运行异步任务，并提供了丰富的操作方法，极大地简化了异步编程。

2.2 CompletableFuture 的基本操作

CompletableFuture 是 Java 提供的一种灵活的工具，它有多种方式可以创建和操作异步任务。

2.2.1 创建 CompletableFuture

以下是一些常见的创建方式：

  1. 直接创建一个空的 CompletableFuture

    CompletableFuture<String> future = new CompletableFuture<>();
// 手动完成任务
future.complete("Task completed");

  2. 使用 supplyAsync 提交异步任务

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {return "Hello, CompletableFuture!";
});

  3. 使用 runAsync 提交无返回值的任务

    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {System.out.println("Task running asynchronously");
});
2.2.2 基本操作 API

创建任务后，CompletableFuture 提供了一系列操作方法来处理结果和任务链：

  1. thenApply：对任务结果进行转换

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Java").thenApply(result -> result + " CompletableFuture");
System.out.println(future.join()); // 输出 "Java CompletableFuture"

  2. thenAccept：消费任务结果，无返回值

    CompletableFuture.supplyAsync(() -> "Hello").thenAccept(result -> System.out.println("Result: " + result));

  3. thenRun：任务完成后执行另一个操作

    CompletableFuture.supplyAsync(() -> "Task Done").thenRun(() -> System.out.println("Next Task Started"));

  4. join 和 get：获取任务结果

    • get() 是阻塞的，可能抛出异常。
    • join() 是非阻塞的，但会抛出 CompletionException。
2.3 ForkJoinPool 与线程池的关系

默认情况下，CompletableFuture 使用 ForkJoinPool.commonPool 作为线程池，这个线程池是一个全局共享的线程池：

  • 适合 CPU 密集型任务：因为它的线程数通常等于 CPU 核心数。
  • 共享资源：多个任务会在同一线程池中运行，可能造成线程争用。
自定义线程池

为了更好地管理线程资源，我们可以为 CompletableFuture 提供自定义线程池：

ExecutorService customExecutor = Executors.newFixedThreadPool(4);CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {return "Using custom executor";
}, customExecutor);

自定义线程池特别适合 I/O 密集型任务，因为它可以通过增加线程数来提高并发能力。

3. 多线程编排常见模式

CompletableFuture 提供了一套强大的 API，使得我们可以优雅地实现多线程任务的编排。以下是一些常见的编排模式及其使用方法。

3.1 串行执行：thenCompose

在某些场景下，任务之间存在依赖关系，例如任务 A 的结果需要传递给任务 B 作为输入。这种情况可以通过 thenCompose 方法实现。

示例：依赖任务的串行执行

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {return "Task A";
}).thenCompose(resultA -> CompletableFuture.supplyAsync(() -> {return resultA + " -> Task B";
}));System.out.println(future.join()); // 输出 "Task A -> Task B"

工作原理：

  • thenCompose 接受一个函数，该函数返回一个新的 CompletableFuture。
  • 当前任务完成后，会将结果传递给该函数，并触发下一个异步任务。
3.2 并行执行：thenCombine 和 allOf

当多个任务可以并行执行，并且需要将它们的结果进行整合时，可以使用以下两种模式。

3.2.1 合并两个任务：thenCombine

thenCombine 用于将两个独立的 CompletableFuture 的结果合并成一个新结果。

示例：两个任务的结果合并

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 5);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 10);CompletableFuture<Integer> result = future1.thenCombine(future2, (num1, num2) -> num1 + num2);System.out.println(result.join()); // 输出 15
3.2.2 合并多个任务：allOf

当有多个任务需要并行执行时，可以使用 allOf 来等待所有任务完成。

示例：等待多个任务完成并聚合结果

List<CompletableFuture<Integer>> futures = Arrays.asList(CompletableFuture.supplyAsync(() -> 1),CompletableFuture.supplyAsync(() -> 2),CompletableFuture.supplyAsync(() -> 3)
);CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));CompletableFuture<List<Integer>> results = allOf.thenApply(v ->futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
);System.out.println(results.join()); // 输出 [1, 2, 3]

注意事项：

  • allOf 本身返回的是 CompletableFuture<Void>，因此需要手动收集子任务的结果。
  • 适用于任务数量较多但彼此独立的场景。
3.3 任意完成：anyOf

在某些情况下，我们只需要最快完成的任务结果，可以使用 anyOf。

示例：获取最快完成的任务结果

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(300);} catch (InterruptedException e) {e.printStackTrace();}return "Task 1 completed";
});CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Task 2 completed");CompletableFuture<Object> fastest = CompletableFuture.anyOf(future1, future2);System.out.println(fastest.join()); // 输出 "Task 2 completed"

应用场景：

  • 多个任务执行时间不同，但只需要最快完成的任务结果。
  • 用于超时机制的实现。
3.4 组合编排示例

场景描述：
一个电商系统中，需要并行获取商品详情、库存信息和用户评价数据，然后整合为一个最终的返回结果。

实现代码：

CompletableFuture<String> productDetails = CompletableFuture.supplyAsync(() -> {return "Product Details";
});CompletableFuture<String> stockInfo = CompletableFuture.supplyAsync(() -> {return "Stock Info";
});CompletableFuture<String> userReviews = CompletableFuture.supplyAsync(() -> {return "User Reviews";
});// 合并结果
CompletableFuture<String> finalResult = productDetails.thenCombine(stockInfo, (details, stock) -> details + " + " + stock).thenCombine(userReviews, (partialResult, reviews) -> partialResult + " + " + reviews);System.out.println(finalResult.join()); // 输出 "Product Details + Stock Info + User Reviews"

4. 实践案例

在实际开发中，CompletableFuture 的多线程编排可以显著优化系统性能和提高代码可维护性。下面通过一个实践案例，演示如何使用 CompletableFuture 来完成复杂的多线程编排任务。

4.1 场景描述

我们以一个电商系统为例，构建一个获取商品详情页信息的服务。该服务需要同时从以下三个数据源获取信息：

  1. 商品基本信息（如名称、描述）。
  2. 库存状态（如库存数量）。
  3. 用户评价（如评分、评论内容）。

要求：

  • 各数据源调用可以并行。
  • 汇总所有结果，返回完整的商品详情。
  • 如果某个数据源调用失败，返回默认值而不是终止整个流程。
4.2 代码实现

以下是基于 CompletableFuture 的实现：

import java.util.concurrent.CompletableFuture;public class ProductService {public static void main(String[] args) {ProductService service = new ProductService();String productDetails = service.getProductDetails("12345");System.out.println(productDetails);}/*** 获取商品详情* @param productId 商品 ID* @return 商品详情*/public String getProductDetails(String productId) {// 异步获取商品基本信息CompletableFuture<String> basicInfoFuture = CompletableFuture.supplyAsync(() -> fetchBasicInfo(productId));// 异步获取库存信息CompletableFuture<String> stockInfoFuture = CompletableFuture.supplyAsync(() -> fetchStockInfo(productId));// 异步获取用户评价信息CompletableFuture<String> userReviewsFuture = CompletableFuture.supplyAsync(() -> fetchUserReviews(productId));// 合并所有结果CompletableFuture<String> productDetailsFuture = basicInfoFuture.thenCombine(stockInfoFuture, (basicInfo, stockInfo) -> basicInfo + "\n" + stockInfo).thenCombine(userReviewsFuture, (partialResult, userReviews) -> partialResult + "\n" + userReviews).exceptionally(ex -> {// 如果发生异常，返回默认信息System.err.println("Error occurred: " + ex.getMessage());return "Error fetching product details";});// 等待结果return productDetailsFuture.join();}// 模拟获取商品基本信息private String fetchBasicInfo(String productId) {simulateDelay(500); // 模拟 500ms 的延迟return "Basic Info: Product Name - Example Product";}// 模拟获取库存信息private String fetchStockInfo(String productId) {simulateDelay(300); // 模拟 300ms 的延迟return "Stock Info: Available - 25 units";}// 模拟获取用户评价private String fetchUserReviews(String productId) {simulateDelay(700); // 模拟 700ms 的延迟return "User Reviews: 4.5/5 - Excellent product!";}// 模拟延迟private void simulateDelay(long millis) {try {Thread.sleep(millis);} catch (InterruptedException e) {e.printStackTrace();}}
}
4.3 代码解读

  1. 并行调用三个异步任务：

    • 每个数据源通过 CompletableFuture.supplyAsync 提交为异步任务。
    • 任务之间互不依赖，因此可以并行运行。

  2. 合并任务结果：

    • 使用 thenCombine 合并多个任务的结果。
    • thenCombine 按顺序两两合并，最终得到完整的商品详情。

  3. 异常处理：

    • 通过 exceptionally 捕获执行过程中可能出现的异常。
    • 如果某个数据源调用失败，返回默认值而不中断整个流程。

  4. 同步等待结果：

    • 使用 join 方法等待所有任务完成，并返回最终结果。
4.4 输出示例

假设所有服务调用正常运行，程序输出如下：

Basic Info: Product Name - Example Product
Stock Info: Available - 25 units
User Reviews: 4.5/5 - Excellent product!

如果某个服务调用失败，例如库存服务抛出异常，输出可能如下：

Basic Info: Product Name - Example Product
Error fetching product details
User Reviews: 4.5/5 - Excellent product!
4.5 性能分析

通过异步编排实现，原本串行执行需要的时间为：

500ms (基本信息) + 300ms (库存信息) + 700ms (用户评价) = 1500ms

使用 CompletableFuture 并行执行后，整体时间由最慢的任务决定，即：

max(500ms, 300ms, 700ms) = 700ms

性能提升明显，特别是在涉及多个独立任务的场景中。

5. 异常处理

在异步编程中，异常处理是不可忽视的部分。CompletableFuture 提供了多种方式来处理任务执行过程中可能出现的异常，确保程序的健壮性和容错性。

5.1 异步任务中的异常传播

在 CompletableFuture 中，如果任务执行时抛出异常，这个异常会被包装成 CompletionException，并传播到任务链的后续节点。例如：

示例：异常传播

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Something went wrong");
});try {future.join(); // join() 抛出 CompletionException
} catch (Exception ex) {System.err.println("Exception: " + ex.getCause().getMessage());
}
// 输出：Exception: Something went wrong
5.2 捕获异常的方式

CompletableFuture 提供了多种方法来捕获和处理异常：

5.2.1 exceptionally

exceptionally 方法允许你捕获异常并返回一个默认值，以便继续执行任务链。

示例：返回默认值

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {if (true) throw new RuntimeException("Error in task");return "Task completed";
}).exceptionally(ex -> {System.err.println("Handled exception: " + ex.getMessage());return "Default Value";
});System.out.println(future.join());
// 输出：Handled exception: java.lang.RuntimeException: Error in task
// 输出：Default Value
5.2.2 handle

handle 方法不仅可以捕获异常，还能处理正常结果。无论任务是否成功，都会执行。

示例：统一处理成功和异常结果

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {if (true) throw new RuntimeException("Error in task");return "Task completed";
}).handle((result, ex) -> {if (ex != null) {System.err.println("Handled exception: " + ex.getMessage());return "Recovered from error";}return result;
});System.out.println(future.join());
// 输出：Handled exception: java.lang.RuntimeException: Error in task
// 输出：Recovered from error
5.2.3 whenComplete

whenComplete 是一种类似“回调”的机制，用于在任务完成后执行某些操作。它不会影响任务的结果，但可以记录日志或执行清理操作。

示例：记录任务状态

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {if (true) throw new RuntimeException("Error in task");return "Task completed";
}).whenComplete((result, ex) -> {if (ex != null) {System.err.println("Task failed with exception: " + ex.getMessage());} else {System.out.println("Task completed successfully with result: " + result);}
});try {future.join();
} catch (Exception ex) {System.err.println("Caught exception: " + ex.getCause().getMessage());
}
// 输出：Task failed with exception: java.lang.RuntimeException: Error in task
// 输出：Caught exception: Error in task
5.3 异常处理的实际场景
5.3.1 容错机制

在微服务调用中，如果某个服务失败，不应终止整个流程，可以通过 exceptionally 或 handle 返回一个默认值。

示例：微服务调用容错

CompletableFuture<String> serviceCall = CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Service unavailable");
}).exceptionally(ex -> "Fallback: Default Service Data");System.out.println(serviceCall.join());
// 输出：Fallback: Default Service Data
5.3.2 记录错误日志

使用 whenComplete 记录任务中的异常信息，便于问题追踪。

示例：记录错误日志

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Error occurred");
}).whenComplete((result, ex) -> {if (ex != null) {System.err.println("Error logged: " + ex.getMessage());}
});future.join();
// 输出：Error logged: java.lang.RuntimeException: Error occurred
5.3.3 局部恢复

对于任务链中间节点的异常，可以通过 handle 在局部进行恢复，而不影响后续任务执行。

示例：局部恢复

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Task failed");
}).handle((result, ex) -> {if (ex != null) return "Recovered Value";return result;
}).thenApply(result -> result + " -> Next Task");System.out.println(future.join());
// 输出：Recovered Value -> Next Task
5.4 异常处理的最佳实践

  1. 使用统一的异常处理工具：
    在复杂的任务链中，最好集中处理异常逻辑，保持代码的简洁性。

  2. 合理设置默认值：
    异常处理时返回的默认值应与业务场景匹配，避免影响系统行为。

  3. 记录日志：
    异常发生时，详细记录日志，便于后续排查问题。

  4. 分层次处理异常：
    在任务链的不同阶段，可以根据需要选择 exceptionally、handle 或 whenComplete，确保异常被合理捕获和处理。

6. 性能优化

在使用 CompletableFuture 进行多线程编排时，性能优化是一个重要的课题。尽管 CompletableFuture 提供了强大的异步处理能力，但不合理的使用可能会导致线程池耗尽、性能下降甚至死锁等问题。本节将讨论如何在实际项目中对 CompletableFuture 进行性能优化。

6.1 自定义线程池的使用

默认情况下，CompletableFuture 使用的是 ForkJoinPool.commonPool 线程池。这个线程池的线程数等于 CPU 核心数，因此适合 CPU 密集型任务。但对于 I/O 密集型任务或需要高度并发的场景，默认线程池可能不足以满足需求。

使用自定义线程池

可以为 CompletableFuture 提供自定义线程池，确保线程资源满足任务需求：

ExecutorService customExecutor = Executors.newFixedThreadPool(10);CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {return "Task result";
}, customExecutor);System.out.println(future.join());
customExecutor.shutdown();

注意事项：

  • 根据任务类型（CPU 密集型或 I/O 密集型）选择线程池的大小。
  • 使用 shutdown() 方法及时释放线程池资源，避免资源泄漏。
6.2 避免阻塞操作

在异步编程中，阻塞操作会降低性能，甚至可能导致线程池耗尽。常见的阻塞操作包括：

  • 使用 Thread.sleep 模拟延迟。
  • 调用同步方法或 I/O 操作。
优化方法

将阻塞操作替换为非阻塞实现。例如，使用异步 I/O 替代传统的同步 I/O：

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {// 模拟 I/O 操作Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}
}).thenRun(() -> System.out.println("Non-blocking task completed"));
future.join();

优化为非阻塞的异步操作：

CompletableFuture.runAsync(() -> {// 模拟非阻塞 I/O 操作System.out.println("Performing non-blocking I/O");
}).join();
6.3 减少线程池争用

当系统中有多个 CompletableFuture 实例同时运行时，共享同一个线程池可能导致线程争用，影响性能。为不同的任务组创建独立的线程池可以有效缓解这一问题。

示例：为不同任务组分配独立线程池

ExecutorService ioExecutor = Executors.newCachedThreadPool();
ExecutorService cpuExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());CompletableFuture<Void> ioTask = CompletableFuture.runAsync(() -> {System.out.println("I/O Task");
}, ioExecutor);CompletableFuture<Void> cpuTask = CompletableFuture.runAsync(() -> {System.out.println("CPU Task");
}, cpuExecutor);CompletableFuture.allOf(ioTask, cpuTask).join();
ioExecutor.shutdown();
cpuExecutor.shutdown();
6.4 合理拆分任务

在并行处理任务时，合理拆分任务可以提高线程利用率，避免线程因任务过大而被长期占用。

示例：拆分大任务为多个小任务
List<Integer> numbers = IntStream.rangeClosed(1, 100).boxed().collect(Collectors.toList());List<CompletableFuture<Integer>> futures = numbers.stream().map(num -> CompletableFuture.supplyAsync(() -> {// 处理单个任务return num * 2;})).collect(Collectors.toList());CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));CompletableFuture<List<Integer>> results = allOf.thenApply(v ->futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
);System.out.println(results.join());
6.5 合并结果时避免重复等待

在使用 CompletableFuture.allOf 或 CompletableFuture.anyOf 时，直接等待所有任务完成可能会导致性能浪费。可以通过提前处理已完成的任务来优化性能。

示例：处理部分已完成的任务

List<CompletableFuture<Integer>> futures = List.of(CompletableFuture.supplyAsync(() -> 1),CompletableFuture.supplyAsync(() -> 2),CompletableFuture.supplyAsync(() -> 3)
);futures.forEach(future -> future.thenAccept(result -> {System.out.println("Processed result: " + result);
}));CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
6.6 性能监控与调优

性能监控和调优是确保 CompletableFuture 高效运行的重要步骤。

常用工具
  1. 线程池监控：使用 JMX 或 APM 工具监控线程池状态（如线程数、队列长度）。
  2. JVM 调优：使用 jvisualvm 或 jconsole 监控 GC、堆内存使用等性能指标。
  3. 日志追踪：为每个异步任务记录执行时间和异常信息，便于分析性能瓶颈。
调优策略
  • 合理调整线程池大小，确保既不过载也不浪费资源。
  • 根据任务类型和耗时动态调整任务分配策略。
  • 优化长时间运行的任务，避免单个任务占用线程池过久。

7. 注意事项

尽管 CompletableFuture 提供了强大的异步编程能力，但在实际使用中也存在一些潜在的陷阱和注意事项。了解这些问题能够帮助开发者更高效、安全地使用 CompletableFuture。

7.1 死锁风险与线程池耗尽
问题描述
  • 当使用 CompletableFuture 的默认线程池 (ForkJoinPool.commonPool) 时，如果任务中包含阻塞操作（如 join() 或 I/O 操作），可能会导致线程耗尽甚至死锁。
  • 特别是在递归任务场景中，线程池耗尽的风险更高。
解决方法

  1. 避免阻塞操作：

    • 尽量不要在异步任务中调用 join() 或其他可能阻塞的操作。
    • 如果需要等待结果，使用 thenApply 等非阻塞方式。

  2. 使用自定义线程池：

    • 为 I/O 密集型任务配置独立线程池，避免默认线程池资源被阻塞任务占用。
    ExecutorService customExecutor = Executors.newCachedThreadPool();
CompletableFuture.runAsync(() -> {// 非阻塞任务
}, customExecutor);
7.2 任务过多对 CPU 的影响
问题描述
  • 当提交的任务数量过多时，会显著增加线程切换的开销，反而导致性能下降。
  • 默认的 ForkJoinPool 使用的线程数与 CPU 核心数一致，过多任务可能无法充分利用并行优势。
解决方法

  1. 优化任务粒度：

    • 将过小的任务合并为更大的任务，减少线程切换的开销。
    CompletableFuture.supplyAsync(() -> {// 处理一个较大的任务，而不是多个小任务return IntStream.range(1, 100).sum();
});

  2. 合理规划线程池大小：

    • 对于 CPU 密集型任务，线程数建议设为 CPU 核心数 + 1。
    • 对于 I/O 密集型任务，线程数可以稍高于任务数量。
7.3 任务链中的异常传播
问题描述
  • 如果任务链中的某个任务发生异常，且未捕获异常，会导致整个链条终止。
  • 异常信息通常被包装为 CompletionException，可能不易发现问题根源。
解决方法

  1. 使用 exceptionally 或 handle 捕获异常：

    • 为每个任务节点添加异常处理逻辑，避免未捕获的异常中断流程。
    CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Error");
}).exceptionally(ex -> {System.err.println("Handled exception: " + ex.getMessage());return "Fallback result";
});

  2. 集中处理异常：

    • 在任务链的末尾集中处理所有未捕获的异常。
    future.exceptionally(ex -> {System.err.println("Unhandled exception: " + ex.getMessage());return null;
});
7.4 注意任务的线程安全性
问题描述
  • 如果多个任务并行访问共享资源（如全局变量或数据结构），可能会导致数据竞争或状态不一致。
解决方法

  1. 避免共享可变状态：

    • 尽量使用线程安全的数据结构（如 ConcurrentHashMap）或设计无状态任务。
    Map<String, String> resultMap = new ConcurrentHashMap<>();
CompletableFuture.runAsync(() -> resultMap.put("key", "value"));

  2. 使用同步机制：

    • 在必要时使用 synchronized 或其他同步工具（如 ReentrantLock）保护共享资源。
    synchronized (sharedResource) {// 安全访问共享资源
}
7.5 防止任务泄漏
问题描述
  • 未及时释放线程池资源可能导致内存泄漏。
  • 某些未完成或被中断的任务可能会占用资源，影响系统性能。
解决方法

  1. 及时关闭线程池：

    • 使用自定义线程池时，确保在程序结束时调用 shutdown() 释放资源。
    customExecutor.shutdown();

  2. 检查未完成任务：

    • 使用 CompletableFuture.isDone() 检查任务状态，并采取相应的清理措施。
7.6 注意依赖链的复杂性
问题描述
  • 复杂的依赖关系可能导致任务链难以维护和调试。
  • 过深的任务链会增加代码复杂性和执行延迟。
解决方法

  1. 简化依赖关系：

    • 将复杂的依赖关系拆分为多个独立模块，分别实现。
    CompletableFuture<String> taskA = CompletableFuture.supplyAsync(() -> "Task A");
CompletableFuture<String> taskB = taskA.thenApply(result -> result + " -> Task B");

  2. 使用注释或日志记录：

    • 在任务链的关键节点添加注释或日志，便于调试和理解流程。
    task.thenAccept(result -> System.out.println("Task completed: " + result));
7.7 避免 join 的过度使用
问题描述
  • 在任务链中频繁调用 join 可能导致性能下降，因为它会阻塞当前线程。
  • 尤其在任务嵌套中使用 join，可能会导致线程耗尽。
解决方法

  1. 使用非阻塞的 thenApply 等方法替代 join：

    future.thenApply(result -> {System.out.println("Result: " + result);return result + " processed";
});

  2. 只在必要时使用 join：

    • 仅在最后需要结果时调用 join，而非任务链中间节点。

8. 扩展与替代方案

CompletableFuture 在 Java 异步编程中发挥着重要作用，但它并非唯一的工具。在某些场景中，其他框架可能提供更丰富的功能或更高的开发效率。本节将探讨 CompletableFuture 的扩展功能及一些常见的替代方案。

8.1 与其他异步工具的对比
8.1.1 CompletableFuture 的优势
  • 标准库支持：无需额外依赖，直接使用 JDK 提供的功能。
  • 简单直观：通过链式 API 实现常见的异步操作和编排。
  • 良好的性能：基于 ForkJoinPool 提供高效的并行执行。
8.1.2 常见替代方案

  1. RxJava

    • 特点：基于观察者模式的响应式编程框架，支持更复杂的异步流处理。
    • 优势：
      • 丰富的操作符（如 map、flatMap 等）。
      • 更好的流处理能力。
    • 适用场景：
      • 需要处理复杂的事件流或数据流。
      • 开发响应式应用程序。
    • 示例代码：
      Observable<String> observable = Observable.just("Hello").map(value -> value + " RxJava");
observable.subscribe(System.out::println);

  2. Project Reactor

    • 特点：基于响应式流规范（Reactive Streams）的框架，是 Spring WebFlux 的核心依赖。
    • 优势：
      • 完全支持非阻塞和背压机制。
      • 与 Spring 集成良好。
    • 适用场景：
      • 开发微服务，使用 Spring WebFlux 或响应式数据操作（如 MongoDB Reactive）。
    • 示例代码：
      Mono<String> mono = Mono.just("Hello").map(value -> value + " Reactor");
mono.subscribe(System.out::println);

  3. Akka

    • 特点：基于 Actor 模型的并发框架，专注于消息驱动的并发处理。
    • 优势：
      • 易于构建高并发、分布式系统。
      • 支持故障隔离和自愈机制。
    • 适用场景：
      • 高度分布式或需要强故障恢复能力的系统。

  4. Kotlin Coroutines

    • 特点：Kotlin 原生协程支持，简化异步代码的编写。
    • 优势：
      • 语法简洁，避免回调地狱。
      • 更接近同步代码的风格。
    • 适用场景：
      • 使用 Kotlin 语言开发的异步任务。
    • 示例代码：
      GlobalScope.launch {val result = async { fetchData() }println(result.await())
}
8.2 CompletableFuture 的扩展
8.2.1 使用第三方库增强功能

  1. Javaslang（Vavr）

    • Vavr 提供了一套函数式编程工具，增强了 Java 的语言能力。
    • 与 CompletableFuture 集成：
      Future<String> future = Future.of(() -> "Hello Vavr");
future.map(value -> value + " CompletableFuture").onSuccess(System.out::println);

  2. Guava ListenableFuture

    • Guava 提供了 ListenableFuture，支持类似 CompletableFuture 的异步处理。
    • 示例代码：
      ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
ListenableFuture<String> future = service.submit(() -> "Hello Guava");
Futures.addCallback(future, new FutureCallback<>() {public void onSuccess(String result) {System.out.println(result);}public void onFailure(Throwable t) {t.printStackTrace();}
}, service);
8.2.2 与异步框架结合

CompletableFuture 可以与其他异步框架或工具结合，提供更强大的能力。例如：

  • 与 Spring Async 集成，利用 Spring 的异步支持。
  • 与 Vert.x 集成，用于构建反应式微服务。
8.3 新版本 Java 的潜在影响
8.3.1 Project Loom

Java 引入的 Project Loom 为并发编程带来了革命性的变化，它引入了轻量级线程（虚拟线程），能够显著简化并发编程的复杂性。

对比 CompletableFuture：

  • CompletableFuture 依赖线程池实现并发，任务数过多时可能导致线程池耗尽。
  • 虚拟线程不依赖操作系统线程，能高效地处理大量任务。

示例代码：使用虚拟线程

ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
executor.submit(() -> {System.out.println("Running in a virtual thread");
});
executor.shutdown();

影响：

  • 在虚拟线程普及后，许多场景下可以直接使用同步编程模型代替 CompletableFuture。
  • 异步编程将变得更加简洁和直观。
8.4 如何选择合适的工具
特性CompletableFutureRxJavaProject ReactorAkkaKotlin Coroutines
标准库支持✅❌❌❌❌
复杂任务编排✅✅✅✅✅
事件流处理❌✅✅❌❌
分布式/高并发支持❌❌❌✅❌
代码简洁性中等中等中等中等高

根据项目需求选择适合的工具：

  • 如果追求轻量级和标准化，优先选择 CompletableFuture。
  • 如果需要流处理和响应式编程，考虑 RxJava 或 Project Reactor。
  • 如果系统是消息驱动的分布式架构，Akka 是首选。
  • 如果使用 Kotlin 开发，推荐 Kotlin Coroutines。

9. 总结

通过本篇文章，我们全面了解了 CompletableFuture 的使用方法和最佳实践，从基础知识到多线程编排，再到异常处理与性能优化，以及对扩展工具和替代方案的探讨。在这里，我们对核心内容进行总结，并提炼关键经验。

9.1 核心价值

  1. 简化异步编程：
    CompletableFuture 提供了一套丰富的 API，支持链式调用和任务编排，让复杂的异步操作更加直观和易读。

  2. 提高系统性能：
    通过任务并行化和非阻塞机制，CompletableFuture 可以显著减少执行时间，特别是在高并发场景中。

  3. 增强容错能力：
    内置的异常处理功能（如 exceptionally 和 handle）使得异步任务的失败不会中断整个流程，从而提高系统的健壮性。

  4. 灵活的任务编排：
    无论是串行、并行还是依赖任务的复杂组合，CompletableFuture 都能提供优雅的解决方案。

9.2 实践经验与最佳实践

  1. 合理使用线程池：

    • 默认线程池适合 CPU 密集型任务，I/O 密集型任务应使用自定义线程池。
    • 避免阻塞操作，以免线程池耗尽。

  2. 任务编排模式：

    • 串行执行：使用 thenCompose，实现任务依赖。
    • 并行执行：使用 thenCombine 或 allOf，实现结果合并。
    • 任意完成：使用 anyOf，优先获取最快完成的任务结果。

  3. 异常处理：

    • 使用 exceptionally 或 handle 捕获异常，提供默认值或恢复逻辑。
    • 在任务链末尾集中处理未捕获的异常。

  4. 性能优化：

    • 拆分大任务，提高并行效率。
    • 减少线程切换开销，优化线程池配置。

  5. 代码可维护性：

    • 使用日志和注释记录关键任务节点，便于调试和排查问题。
    • 对复杂任务链分层设计，降低耦合度。
9.3 拓展与未来展望

  1. 扩展使用场景：

    • 在微服务架构中，CompletableFuture 可以优化服务间调用的并行性。
    • 在批量数据处理场景中，利用其强大的并行编排能力提升性能。

  2. 结合其他工具：

    • 如果项目对事件流或响应式编程有需求，可以结合 RxJava 或 Project Reactor。
    • 使用 Kotlin 时，可优先考虑 Coroutines，其语法更加简洁。

  3. Project Loom 的影响：

    • 虚拟线程的引入可能会改变并发编程的传统方式。
    • CompletableFuture 在 Loom 环境下可能与虚拟线程结合，进一步提升性能和简化异步任务管理。
9.4 适用场景总结
场景推荐工具原因
简单的异步任务或多线程编排CompletableFuture简单易用，支持标准库，减少依赖
复杂事件流或响应式数据处理RxJava 或 Project Reactor更丰富的操作符，适合复杂数据流场景
分布式系统中的消息驱动任务Akka基于 Actor 模型，支持高并发和分布式
Kotlin 开发的异步任务Kotlin Coroutines语法简洁，代码风格接近同步逻辑
高性能、高并发任务（未来发展）Java Loom + CompletableFuture虚拟线程结合 CompletableFuture 提供更高性能和简洁性

10. 附录

为了便于读者快速掌握和实践 CompletableFuture 的核心功能，本附录提供常用代码片段、参考资料以及完整的示例代码，帮助开发者更高效地学习和使用 CompletableFuture。

10.1 常用代码片段

  1. 创建异步任务

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello, CompletableFuture");

  2. 任务依赖（串行执行）

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Task A").thenCompose(resultA -> CompletableFuture.supplyAsync(() -> resultA + " -> Task B"));
System.out.println(future.join());

  3. 并行任务

    • 合并两个任务

      CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 5).thenCombine(CompletableFuture.supplyAsync(() -> 10), (a, b) -> a + b);
System.out.println(future.join()); // 输出 15

    • 等待多个任务完成

      CompletableFuture<Void> allOf = CompletableFuture.allOf(CompletableFuture.runAsync(() -> System.out.println("Task 1")),CompletableFuture.runAsync(() -> System.out.println("Task 2"))
);
allOf.join();

  4. 异常处理

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Error occurred");
}).exceptionally(ex -> "Default Value");
System.out.println(future.join());

  5. 使用自定义线程池

    ExecutorService customExecutor = Executors.newFixedThreadPool(4);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Using custom executor", customExecutor);
System.out.println(future.join());
customExecutor.shutdown();
10.2 示例：完整代码

场景：从三个服务获取数据（商品详情、库存信息和用户评价），并合并结果返回。

import java.util.concurrent.CompletableFuture;public class ProductDetailsService {public static void main(String[] args) {ProductDetailsService service = new ProductDetailsService();String result = service.getProductDetails("12345");System.out.println(result);}public String getProductDetails(String productId) {CompletableFuture<String> productInfoFuture = CompletableFuture.supplyAsync(() -> fetchProductInfo(productId));CompletableFuture<String> stockInfoFuture = CompletableFuture.supplyAsync(() -> fetchStockInfo(productId));CompletableFuture<String> userReviewsFuture = CompletableFuture.supplyAsync(() -> fetchUserReviews(productId));CompletableFuture<String> combinedFuture = productInfoFuture.thenCombine(stockInfoFuture, (productInfo, stockInfo) -> productInfo + "\n" + stockInfo).thenCombine(userReviewsFuture, (partialResult, userReviews) -> partialResult + "\n" + userReviews).exceptionally(ex -> "Error fetching product details");return combinedFuture.join();}private String fetchProductInfo(String productId) {simulateDelay(500);return "Product Info: Product ID " + productId;}private String fetchStockInfo(String productId) {simulateDelay(300);return "Stock Info: Available - 20 units";}private String fetchUserReviews(String productId) {simulateDelay(700);return "User Reviews: 4.8/5 - Excellent!";}private void simulateDelay(long millis) {try {Thread.sleep(millis);} catch (InterruptedException e) {e.printStackTrace();}}
}

运行结果：

Product Info: Product ID 12345
Stock Info: Available - 20 units
User Reviews: 4.8/5 - Excellent!
10.3 参考资料

  1. 官方文档

    • Java CompletableFuture API 文档

  2. 深入学习

    • 《Java 并发编程实战》：经典并发编程书籍。
    • 《Effective Java》：提供关于并发和多线程的最佳实践。

  3. 相关博客和文章

    • CompletableFuture 教程（Baeldung）
    • Java 并发工具详解

  4. 示例代码仓库

    • GitHub - CompletableFuture Examples

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.xdnf.cn/news/35870.html

如若内容造成侵权/违法违规/事实不符，请联系一条长河网进行投诉反馈，一经查实，立即删除！

相关文章

百度智能云 CHPC: 使用 BtuneAK对基因测序软件进行加速

百度智能云 CHPC: 使用 BtuneAK对基因测序软件进行加速

背景 本文主要介绍在 CHPC 平台使用 BtuneAK 自动化加速组件&#xff0c;可以直接对BWA、FastQC、Picard、Trimmomatic等业务端到端时长加速。 Btune 简单介绍 BtunePK介绍 BtunePK 是百度自研的一款性能分析和调优工具&#xff0c;兼容Intel、AMD、ARM三个CPU平台&#xff0…
阅读更多...
Power BI - 批量导入数据

Power BI - 批量导入数据

1.简单介绍 假定已经使用Power Automate Desktop(微软的RPA产品&#xff0c;是Power Platform平台的其中一个产品)从福布斯中文网获取了各地区的2024年的财富数据如下&#xff0c; 现在想批量导入数据到Power BI中&#xff0c;分析一下各地区的产业以及财富情况 2.具体说明 …
阅读更多...
实现跨平台 SSH 连接：从 macOS 到 Windows WSL 的完整解决方案20241203

实现跨平台 SSH 连接：从 macOS 到 Windows WSL 的完整解决方案20241203

&#x1f310; 实现跨平台 SSH 连接&#xff1a;从 macOS 到 Windows WSL 的完整解决方案 ✨ 引言 随着跨平台开发的普及&#xff0c;开发者经常需要在多系统环境中切换和协作。尤其是在 macOS 和 Windows 混合使用的开发环境中&#xff0c;通过 SSH 远程访问和管理 Windows …
阅读更多...
【css】基础（二）

【css】基础（二）

本专栏内容为&#xff1a;前端专栏 记录学习前端&#xff0c;分为若干个子专栏&#xff0c;html js css vue等 &#x1f493;博主csdn个人主页&#xff1a;小小unicorn ⏩专栏分类&#xff1a;css专栏 &#x1f69a;代码仓库&#xff1a;小小unicorn的代码仓库&#x1f69a; &a…
阅读更多...
2024通信工程师-中级-互联网技术备考经验

2024通信工程师-中级-互联网技术备考经验

考试简介 全国通信专业技术人员职业水平考试&#xff0c;是由国家人力资源和社会保障部、工业和信息化部领导下的国家级考试。根据原人事部、信息产业部文件&#xff08;国人部发[2006]10号&#xff09;&#xff0c;通信专业技术人员职业水平评价&#xff0c;纳入全国专业技术人…
阅读更多...
智能文档解析综述：结构化信息提取的技术、挑战与前景

智能文档解析综述：结构化信息提取的技术、挑战与前景

综述论文&#xff1a;https://arxiv.org/abs/2410.21169 摘要 文档解析对于将非结构化和半结构化文档&#xff08;如合同、学术论文和发票&#xff09;转换为结构化、机器可读的数据至关重要。通过从非结构化输入中提取可靠的结构化数据&#xff0c;文档解析为众多应用提供了极…
阅读更多...
【Web】AlpacaHack Round 7 (Web) 题解

【Web】AlpacaHack Round 7 (Web) 题解

Treasure Hunt flag在md5值拼接flagtxt的文件里&#xff0c;如 d/4/1/d/8/c/d/9/8/f/0/0/b/2/0/4/e/9/8/0/0/9/9/8/e/c/f/8/4/2/7/e/f/l/a/g/t/x/t 访问已经存在的目录状态码是301 访问不存在的目录状态码是404 基于此差异可以写爆破脚本 这段waf可以用url编码绕过 做个lab …
阅读更多...
操作系统——文件系统

操作系统——文件系统

笔记内容及图片整理自XJTUSE “操作系统” 课程ppt&#xff0c;仅供学习交流使用&#xff0c;谢谢。 文件系统是操作系统中以文件方式管理计算机软件资源的软件和被管理的文件和数据结构&#xff08;如目录和索引表等&#xff09;的集合。从系统角度来看&#xff0c;文件系统是…
阅读更多...
java面向对象实验——扫雷+24点

java面向对象实验——扫雷+24点

扫雷 窗口绘制&#xff1a; GameWin package com.sxt;import javax.swing.*;public class GameWin extends JFrame {void launch(){this.setVisible(true);this.setSize(500, 500);this.setLocationRelativeTo(null);this.setTitle("SWE23070扫雷游戏");this.setD…
阅读更多...
GPU 调度策略架构与CUDA运行机制（二）

GPU 调度策略架构与CUDA运行机制（二）

市面上有很多GPU厂家&#xff0c;他们产品的软硬件架构各不相同&#xff0c;但是核心往往差不多&#xff0c;整明白了一个基本上就可以触类旁通了。针对当前gpu底层的一些架构以及硬件层一些调度策略的话估计大部分人就很难说的上熟悉了&#xff0c;这个不是大家的错&#xff0…
阅读更多...
ddos攻击防御的方法有哪些

ddos攻击防御的方法有哪些

DDoS攻击&#xff0c;即分布式拒绝服务攻击(Distributed Denial of Service)&#xff0c;是一种恶意的网络攻击方式&#xff0c;旨在通过发送大量流量或请求到目标服务器、服务或网络&#xff0c;使其资源耗尽&#xff0c;无法处理合法用户的请求&#xff0c;从而导致服务中断或…
阅读更多...
Python + Playwright：集成 Applitools 进行视觉回归测试（快速入门）

Python + Playwright：集成 Applitools 进行视觉回归测试（快速入门）

集成 Applitools 进行视觉回归测试（快速入门） 简介Applitools 的核心特点Applitools 的应用场景1. 准备工作2. 获取示例项目2.1 下载示例代码2.2 安装依赖2.3 选择测试运行方式3. 代码解析3.1 测试用例示例4. 运行测试4.1 设置 Applitools API 变量4.2 设置 Applitools Eyes …
阅读更多...
RuoYi中数据分页功能实现的步骤（nvliz）

RuoYi中数据分页功能实现的步骤（nvliz）

目录 前言 数据分页的作用 RuoYi中的实现步骤 前端的显示界面(实例介绍) 源码分析&#xff08;前端&#xff09; Pagination&#xff08;分页组件&#xff09;介绍 前端&#xff1a;getList()(方法源码分析) 源码分析&#xff08;后端&#xff09; 后端&#xff1a;List()…
阅读更多...
HarmonyOS 5.0应用开发——全局广播的使用

HarmonyOS 5.0应用开发——全局广播的使用

【高心星出品】 文章目录 全局广播的使用公共事件接受系统公共事件原理 发布与订阅自定义公共事件订阅系统事件 全局广播的使用 全局广播可以用来做应用间通信&#xff0c;进程间通信&#xff0c;包含订阅、发布等功能。 公共事件 CES&#xff08;Common Event Service&…
阅读更多...
ceph存储池

ceph存储池

1、存储池 1、存储池的概念 存储池就是ceph的逻辑分区&#xff0c;专门用来存储对象的 特点 将文件切片成对象&#xff0c;通过hash算法&#xff0c;找到存储池中的pg&#xff0c;池中的pg根据crush算法找到osd节点 存储中的PG数量对性能有重要的影响&#xff0c;过多和过少…
阅读更多...
Ollama记录

Ollama记录

先在官网下载Ollama 模型下载 ollama run qwen2:0.5b 可以快速部署很多模型 方便 可以替换openai api key from openai import OpenAIclient OpenAI(base_url https://127.0.0.1:11434/v1,api_keyollama, # required, but unused )response client.chat.completions.…
阅读更多...
锻造船用发动机动力系统，铸强船舶“心脏”

锻造船用发动机动力系统，铸强船舶“心脏”

船舶是海洋、湖泊及河流中重要的水上交通工具&#xff0c;不仅能够促进海上经济的发展&#xff0c;还能够保卫国家的制海权。船舶动力装置&#xff0c;也就是船舶的核心动力源——船用发动机动力系统对船舶的重要作用不言自明&#xff0c;关系到船舶的性能质量&#xff0c;能够…
阅读更多...
LIN状态管理

LIN状态管理

文章目录 前言一、状态管理二、实现过程三、response_error信号的变化条件四、节点内部报告五、测试应用方式1&#xff1a;LIN ISC方式2&#xff1a;CAPL脚本方式 前言 在LIN专栏的文章中&#xff0c;我们介绍了 LIN的网络管理&#xff1a;LIN网络管理&#xff1a;休眠&唤…
阅读更多...
云帆在线学习考试系统对国产化数据库的支持情况说明

云帆在线学习考试系统对国产化数据库的支持情况说明

云帆在线学习考试系统对国产化数据库的支持情况说明 云帆学习考试系统是一款优秀的学习和考试系统&#xff01;多年以来一直深耕在线教育板块&#xff0c;积累了丰富的行业经验&#xff0c;多年来的产品和技术沉淀&#xff0c;服务了众多政府机构、知名高校、企事业单位。 今…
阅读更多...
反向传播算法中的误差项

反向传播算法中的误差项

背景 在反向传播算法中&#xff0c;我们需要计算每个神经元的误差项&#xff0c;以便更新网络中的权重。对于输出层的神经元&#xff0c;误差项的计算公式如下&#xff1a; 其中&#xff1a; E是损失函数&#xff08;例如均方误差&#xff09;。 zk 是输出层神经元的加权输入&a…
阅读更多...
最新文章

Copyright @ 2022~2023