CompletableFuture 详解
CompletableFuture 详解
在 Java 并发编程领域，CompletableFuture 是一个强大的工具，它不仅实现了 Future 接口，还提供了丰富的组合、链接和处理异步计算结果的功能。本文将深入探讨 CompletableFuture 的核心概念、API 以及实际应用场景。
一、CompletableFuture 概述
1.1 什么是 CompletableFuture
CompletableFuture 是 Java 8 引入的一个类，位于 java.util.concurrent 包下，它实现了 Future 和 CompletionStage 接口。Future 表示一个异步计算的结果，但它提供的功能有限，而 CompletableFuture 则扩展了这些功能，允许我们以声明式的方式处理异步计算结果。
1.2 为什么需要 CompletableFuture
在 Java 8 之前，处理异步计算主要依靠 Future 接口，但 Future 有以下局限性：
- 无法手动完成计算（除非通过
FutureTask的set方法） - 缺乏组合多个
Future的能力 - 没有提供异常处理机制
- 不能注册回调函数
CompletableFuture 解决了这些问题，它提供了约 50 种不同的方法来组合、链接和处理 CompletableFuture 实例的结果。
二、CompletableFuture 基础用法
2.1 创建 CompletableFuture
2.1.1 创建一个已完成的 CompletableFuture
CompletableFuture<String> completedFuture = CompletableFuture.completedFuture("Hello, World!");
String result = completedFuture.get(); // 立即返回 "Hello, World!"2.1.2 创建一个尚未完成的 CompletableFuture
CompletableFuture<String> future = new CompletableFuture<>();
// 稍后在其他线程中完成这个 future
new Thread(() -> {
try {
Thread.sleep(1000);
future.complete("Hello after 1 second");
} catch (InterruptedException e) {
future.completeExceptionally(e);
}
}).start();2.1.3 使用 supplyAsync 创建一个异步执行的 CompletableFuture
// 使用默认线程池（ForkJoinPool.commonPool()）
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "Result from supplyAsync";
});
// 使用自定义线程池
ExecutorService executor = Executors.newFixedThreadPool(5);
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "Result from supplyAsync with custom executor";
}, executor);2.1.4 使用 runAsync 创建一个不返回结果的 CompletableFuture
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(1000);
System.out.println("Task completed");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});2.2 获取结果
2.2.1 使用 get() 方法（可能阻塞）
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
String result = future.get(); // 阻塞直到任务完成2.2.2 使用 join() 方法（不抛出检查型异常）
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (Math.random() > 0.5) {
throw new RuntimeException("Random failure");
}
return "Hello";
});
String result = future.join(); // 阻塞直到任务完成，遇到异常会抛出CompletionException2.2.3 使用 getNow() 获取当前结果或默认值
CompletableFuture<String> future = new CompletableFuture<>();
String result = future.getNow("Default Value"); // 如果future未完成，返回默认值2.2.4 使用 completeOnTimeout 设置超时后完成的值
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "Result after 2 seconds";
});
// 如果在1秒内未完成，则返回默认值
CompletableFuture<String> result = future.completeOnTimeout("Default on timeout", 1, TimeUnit.SECONDS);2.2.5 使用 orTimeout 设置超时异常
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "Result after 2 seconds";
});
// 如果在1秒内未完成，则抛出TimeoutException
CompletableFuture<String> result = future.orTimeout(1, TimeUnit.SECONDS);三、CompletableFuture 组合操作
3.1 转换结果（thenApply、thenApplyAsync）
thenApply 方法用于对 CompletableFuture 的结果进行转换，类似于 Stream 中的 map 操作。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
.thenApply(s -> s + " World")
.thenApply(s -> s.toUpperCase());
// 输出: HELLO WORLD
System.out.println(future.join());thenApplyAsync 与 thenApply 类似，但它会在另一个线程中执行转换操作。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
System.out.println("Supply in thread: " + Thread.currentThread().getName());
return "Hello";
}).thenApplyAsync(s -> {
System.out.println("Apply in thread: " + Thread.currentThread().getName());
return s + " World";
});3.2 消费结果（thenAccept、thenAcceptAsync）
thenAccept 方法用于消费 CompletableFuture 的结果，不返回任何值。
CompletableFuture.supplyAsync(() -> "Hello")
.thenAccept(s -> System.out.println("Received: " + s));3.3 完成后执行（thenRun、thenRunAsync）
thenRun 方法在 CompletableFuture 完成后执行一个 Runnable，但不关心其结果。
CompletableFuture.supplyAsync(() -> {
System.out.println("Task is running");
return "Result";
}).thenRun(() -> System.out.println("Task is completed"));3.4 组合两个 CompletableFuture（thenCompose）
thenCompose 方法用于组合两个依赖的异步操作，类似于 Stream 中的 flatMap 操作。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));3.5 组合两个独立的 CompletableFuture（thenCombine）
thenCombine 方法用于组合两个独立的异步操作，并对它们的结果进行处理。
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");
CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2);
// 输出: Hello World
System.out.println(combinedFuture.join());3.6 等待多个 CompletableFuture 完成（allOf、anyOf）
allOf 方法等待所有给定的 CompletableFuture 完成。
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Result 1");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Result 2");
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "Result 3");
CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(future1, future2, future3);
// 等待所有任务完成
allOfFuture.join();
// 获取各个结果
String result1 = future1.join();
String result2 = future2.join();
String result3 = future3.join();anyOf 方法等待任意一个给定的 CompletableFuture 完成。
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
try { Thread.sleep(300); } catch (InterruptedException e) {}
return "Result 1";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
try { Thread.sleep(200); } catch (InterruptedException e) {}
return "Result 2";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
try { Thread.sleep(100); } catch (InterruptedException e) {}
return "Result 3";
});
CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(future1, future2, future3);
// 输出: Result 3（因为它最快完成）
System.out.println(anyOfFuture.join());四、CompletableFuture 异常处理
4.1 异常处理（exceptionally）
exceptionally 方法用于处理 CompletableFuture 中的异常情况。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (true) {
throw new RuntimeException("Something went wrong");
}
return "Normal result";
}).exceptionally(ex -> {
System.out.println("Caught exception: " + ex.getMessage());
return "Default value";
});
// 输出: Default value
System.out.println(future.join());4.2 处理结果或异常（handle）
handle 方法同时处理正常结果和异常情况。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (Math.random() > 0.5) {
throw new RuntimeException("Random failure");
}
return "Success";
}).handle((result, ex) -> {
if (ex != null) {
System.out.println("Handled exception: " + ex.getMessage());
return "Recovered from failure";
}
return result;
});4.3 执行完成后的处理（whenComplete）
whenComplete 方法在 CompletableFuture 完成后执行，无论是否发生异常。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (Math.random() > 0.5) {
throw new RuntimeException("Random failure");
}
return "Success";
}).whenComplete((result, ex) -> {
if (ex != null) {
System.out.println("Operation failed: " + ex.getMessage());
} else {
System.out.println("Operation succeeded with result: " + result);
}
});五、CompletableFuture 高级用法
5.1 自定义线程池
CompletableFuture 默认使用 ForkJoinPool.commonPool()，但在实际生产环境中，建议使用自定义线程池以避免耗尽公共线程池资源。
ExecutorService customExecutor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
// 异步任务
return "Result";
}, customExecutor);
// 使用完后关闭线程池
customExecutor.shutdown();5.2 并行处理列表
CompletableFuture 非常适合并行处理列表中的元素。
List<String> items = Arrays.asList("item1", "item2", "item3", "item4", "item5");
ExecutorService executor = Executors.newFixedThreadPool(5);
List<CompletableFuture<String>> futures = items.stream()
.map(item -> CompletableFuture.supplyAsync(() -> {
// 处理每个元素
try { Thread.sleep(100); } catch (InterruptedException e) {}
return "Processed " + item;
}, executor))
.collect(Collectors.toList());
// 等待所有任务完成并收集结果
List<String> results = futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList());
executor.shutdown();5.3 异步方法链的构建
CompletableFuture 的一大优势是可以构建复杂的异步方法链。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
System.out.println("Step 1: Prepare data");
return "Data";
}).thenApplyAsync(data -> {
System.out.println("Step 2: Process data");
return "Processed " + data;
}).thenCombineAsync(CompletableFuture.supplyAsync(() -> {
System.out.println("Step 3: Prepare additional data");
return "Additional data";
}), (result1, result2) -> {
System.out.println("Step 4: Combine results");
return result1 + " and " + result2;
}).thenApplyAsync(finalResult -> {
System.out.println("Step 5: Final processing");
return finalResult.toUpperCase();
}).exceptionally(ex -> {
System.out.println("Error occurred: " + ex.getMessage());
return "Default result";
});
System.out.println("Final result: " + future.join());六、CompletableFuture 性能与最佳实践
6.1 性能考虑
使用自定义线程池：避免使用默认的
ForkJoinPool.commonPool()，特别是在高并发环境下。合理设置线程池大小：根据任务性质和系统资源设置合适的线程池大小。
避免阻塞操作：在
CompletableFuture的回调中避免执行阻塞操作，以免占用线程池资源。
6.2 最佳实践
总是处理异常：使用
exceptionally、handle或whenComplete处理异常，避免异常被静默忽略。关闭线程池：使用完自定义线程池后，记得调用
shutdown()方法关闭它。避免过度使用：虽然
CompletableFuture很强大，但不是所有场景都需要使用它。对于简单的异步任务，FutureTask可能已经足够。组合操作时注意依赖关系：使用
thenCompose处理依赖关系，使用thenCombine或allOf处理独立任务。使用
complete()和completeExceptionally()手动控制：在某些场景下，可能需要手动完成CompletableFuture或使其失败。
七、总结
CompletableFuture 是 Java 并发编程中的一个强大工具，它提供了丰富的 API 来处理异步计算结果。通过 CompletableFuture，我们可以以声明式的方式组合、链接和处理异步操作，大大简化了异步编程的复杂性。
在实际应用中，CompletableFuture 常用于：
- 并行处理独立任务
- 构建复杂的异步工作流
- 实现非阻塞的 API
- 处理异步计算结果
掌握 CompletableFuture 的使用，对于提升 Java 并发编程能力和构建高性能应用程序具有重要意义。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来源 lucky845的博客！
评论