CompletableFuture结合Stream可高效并行处理任务集合，通过allOf等待所有任务完成，并用join收集结果；需合理处理异常避免整体中断，推荐自定义线程池提升控制力，同时支持thenApply、thenCompose等链式操作构建异步流水线。
在Java中，CompletableFuture 是实现异步编程的重要工具，特别适合并行处理任务集合。它基于ForkJoinPool，能有效利用线程资源，提升系统吞吐量。当你有一组独立任务需要并发执行，并希望在所有任务完成后统一处理结果时，CompletableFuture提供了简洁且高效的解决方案。
并行执行多个任务并收集结果
使用 CompletableFuture.allOf() 可以等待多个异步任务全部完成。配合 Stream.map() 和 collect()，可以轻松实现任务集合的并行处理。
示例：假设有10个耗时操作（如远程调用或文件读取），我们希望并行执行它们并获取最终结果列表：
import java.util.concurrent.CompletableFuture;
import java.util.List;
import java.util.stream.Collectors;
public class AsyncTaskExample {
public static void main(String[] args) {
List<String> tasks = List.of("task1", "task2", "task3", "task4", "task5");
// 将每个任务转为CompletableFuture<String>
List<CompletableFuture<String>> futures = tasks.stream()
.map(task -> CompletableFuture.supplyAsync(() -> {
// 模拟耗时操作
try { Thread.sleep(1000); } catch (InterruptedException e) {}
return "Processed: " + task;
}))
.collect(Collectors.toList());
// 等待所有任务完成
CompletableFuture<Void> allDone = CompletableFuture.allOf(
futures.toArray(new CompletableFuture[0])
);
// 获取结果列表
CompletableFuture<List<String>> results = allDone.thenApply(v ->
futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList())
);
// 输出结果
results.thenAccept(list -> list.forEach(System.out::println));
// 主线程等待（实际中可用join避免提前退出）
results.join();
}
}异常处理与容错机制
并行任务中某个失败会导致整体中断，因此必须合理处理异常。可以在 supplyAsync 中捕获异常并返回默认值，或使用 exceptionally() 方法兜底。
立即学习“Java免费学习笔记（深入）”；
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (Math.random() < 0.5) throw new RuntimeException("Processing failed");
return "Success";
}).exceptionally(ex -> {
System.err.println("Error: " + ex.getMessage());
return "Fallback Result"; // 出错时返回默认值
});这样即使个别任务失败，也不会影响其他任务和整体流程。
自定义线程池提升控制力
默认情况下，CompletableFuture 使用 ForkJoinPool.commonPool()，但在高并发或特定场景下建议使用自定义线程池，避免阻塞公共线程池。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
ExecutorService executor = Executors.newFixedThreadPool(10);
List<CompletableFuture<String>> futures = tasks.stream()
.map(task -> CompletableFuture.supplyAsync(() -> {
// 业务逻辑
return "Handled by custom pool: " + task;
}, executor))
.collect(Collectors.toList());
// 记得关闭线程池
executor.shutdown();流式异步任务链（Pipeline）
CompletableFuture 支持链式调用，可构建异步流水线。例如：查询 → 处理 → 保存，每步都不阻塞主线程。
CompletableFuture
.supplyAsync(() -> fetchUserData()) // 第一步：获取数据
.thenApplyAsync(data -> enrichData(data)) // 第二步：加工数据
.thenCompose(data -> saveAsync(data)) // 第三步：保存（返回CompletableFuture）
.thenAccept(result -> log("Saved: " + result));注意：thenCompose 用于扁平化嵌套的 CompletableFuture，类似于flatMap。
基本上就这些。CompletableFuture 结合 Stream 能高效处理任务集合，关键是理解 allOf、join、异常处理和线程池配置。不复杂但容易忽略细节。
