深入掌握 Java CompletableFuture 异步编程
异步任务的创建
在 Java 中，CompletableFuture 提供了灵活的方式来开启异步线程。主要分为无返回值的 runAsync 和有返回值的 supplyAsync。
1. supplyAsync：支持返回值的异步任务
当你需要异步执行一个操作并获取其结果时，应使用此方法。它可以配合默认的 ForkJoinPool 或自定义线程池使用。
// 使用默认线程池
CompletableFuture<String> fetchFuture = CompletableFuture.supplyAsync(() -> {
// 模拟耗时操作
return "Data from Remote Service";
});
// 使用自定义线程池
Executor executor = Executors.newFixedThreadPool(4);
CompletableFuture<Integer> calculationFuture = CompletableFuture.supplyAsync(() -> {
return 42;
}, executor);
2. runAsync：无返回值的异步任务
适用于只需要触发某个异步操作而不需要关心其执行结果的场景。
CompletableFuture<Void> logFuture = CompletableFuture.runAsync(() -> {
System.out.println("异步日志记录中...");
});
任务回调与链式操作
1. thenRun 与 thenRunAsync
在当前任务完成后执行下一步，不接收参数也不返回结果。thenRun 通常在当前线程执行，而 thenRunAsync 会提交到线程池异步执行。
CompletableFuture.supplyAsync(() -> "Task 1")
.thenRun(() -> System.out.println("Task 1 已完成，开始执行清理工作"));
2. thenAccept 与 thenAcceptAsync
接收上一个任务的输出作为输入，但本身没有返回值。
CompletableFuture.supplyAsync(() -> "Result A")
.thenAccept(result -> System.out.println("接收到结果: " + result));
3. thenApply 与 thenApplyAsync
最常用的转换操作，接收上一步的结果，处理后返回一个新的结果。
CompletableFuture<Integer> lengthFuture = CompletableFuture.supplyAsync(() -> "Hello World")
.thenApply(text -> text.length());
异常处理与结果收尾
1. exceptionally
用于捕捉链式调用中抛出的任何异常。如果发生异常，可以通过此方法提供一个默认值或处理逻辑。
CompletableFuture<String> safeFuture = CompletableFuture.supplyAsync(() -> {
if (true) throw new RuntimeException("Service Failure");
return "Success";
}).exceptionally(ex -> {
System.err.println("错误信息: " + ex.getMessage());
return "Fallback Data";
});
2. whenComplete 与 handle
- whenComplete：在任务完成后触发，接收结果和异常，但无法改变返回结果（返回的仍是上一步的 Future）。
- handle：功能更强大，接收结果和异常，并且可以返回一个新的转换结果。
CompletableFuture<String> resultFuture = CompletableFuture.supplyAsync(() -> "Process")
.handle((res, ex) -> {
if (ex != null) return "Error Occurred";
return res + " Finished";
});
多任务组合处理
AND 组合关系
只有当两个任务都完成时，才触发后续动作。主要包括 thenCombine（有输入有输出）、thenAcceptBoth（有输入无输出）和 runAfterBoth（无输入无输出）。
CompletableFuture<String> taskA = CompletableFuture.supplyAsync(() -> "Service A");
CompletableFuture<String> taskB = CompletableFuture.supplyAsync(() -> "Service B");
CompletableFuture<String> combined = taskA.thenCombine(taskB, (resA, resB) -> resA + " & " + resB);
OR 组合关系
只要其中一个任务先完成，就执行后续动作。例如 applyToEither 或 acceptEither。
CompletableFuture<String> fastTask = CompletableFuture.supplyAsync(() -> "Fast Result");
CompletableFuture<String> slowTask = CompletableFuture.supplyAsync(() -> {
Thread.sleep(1000);
return "Slow Result";
});
fastTask.acceptEither(slowTask, result -> System.out.println("最快的结果是: " + result));
allOf 与 anyOf
- allOf：等待所有 Future 完成。
- anyOf：只要有一个 Future 完成即返回。
CompletableFuture.allOf(f1, f2, f3).join(); // 阻塞直到所有任务结束
最佳实践与注意事项
1. 警惕阻塞操作
get() 方法是阻塞的。在生产环境中，应始终使用带超时时间的 get(long timeout, TimeUnit unit)，以防线程永久挂起。
2. 异常屏蔽
如果在异步任务中发生异常且没有显式调用 get() 或处理异常回调，异常可能会被"吞掉"。务必通过 exceptionally 或在获取结果时处理 ExecutionException。
3. 线程池配置
默认的 ForkJoinPool.commonPool() 线程数较少（通常为 CPU 核数 - 1）。对于 IO 密集型任务，建议使用自定义 ThreadPoolExecutor，并合理配置拒绝策略（如 AbortPolicy），避免任务静默丢弃导致无法触发回调。
4. thenCompose 与 thenApply 的区别
如果回调函数本身也返回一个 CompletableFuture，使用 thenApply 会产生嵌套结构（Future 内嵌 Future），而 thenCompose 会将其"扁平化"。
// thenCompose 会展开内部的 Future
CompletableFuture<String> finalResult = firstTask.thenCompose(data -> getSecondTask(data));
