CompletableFuture的5大坑！-智慧文博士

前言

CompletableFuture在并发编程中非常实用，但如果用不好，也很容易踩坑。

今天这篇文章跟大家一起聊聊，CompletableFuture在使用过程中最常见的那些坑，希望对你会有所帮助。

一、CompletableFuture简介

有些小伙伴在工作中刚开始接触CompletableFuture时，可能会被它强大的功能所吸引。

确实，CompletableFuture为我们提供了非常优雅的异步编程方式，但正如武侠小说中的神兵利器，如果使用不当，反而会伤到自己。

CompletableFuture的基本用法

先来看一个简单的CompletableFuture使用示例：

public class BasicCompletableFutureDemo { public static void main(String[] args) throws Exception { // 简单的异步计算 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { // 模拟耗时操作 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } return "Hello, CompletableFuture!"; }); // 获取结果（阻塞） String result = future.get(); System.out.println(result); } }

看起来很简单对吧？但正是这种表面上的简单，掩盖了很多潜在的复杂性。

让我们通过一个架构图来理解CompletableFuture的完整生态：

现在，让我们开始深入探讨各个坑点。

二、线程池使用不当

有些小伙伴在使用CompletableFuture时，往往忽略了线程池的配置，这可能是最容易被忽视但影响最大的坑。

默认线程池的陷阱

public class ThreadPoolPitfall { // 危险的用法：大量使用默认线程池 public void processBatchData(List<String> dataList) { List<CompletableFuture<String>> futures = new ArrayList<>(); for (String data : dataList) { // 使用默认的ForkJoinPool.commonPool() CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { return processData(data); }); futures.add(future); } // 等待所有任务完成 CompletableFuture.allOf(fatures.toArray(new CompletableFuture[0])) .join(); } private String processData(String data) { // 模拟数据处理 try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return data.toUpperCase(); } }

问题分析：

默认线程池大小是CPU核心数-1
在IO密集型任务中，这会导致大量任务排队等待
如果任务提交速度 > 任务处理速度，会造成内存溢出

正确的线程池使用方式

public class ProperThreadPoolUsage { private final ExecutorService ioBoundExecutor; private final ExecutorService cpuBoundExecutor; public ProperThreadPoolUsage() { // IO密集型任务 - 使用较大的线程池 this.ioBoundExecutor = new ThreadPoolExecutor( 50, // 核心线程数 100, // 最大线程数 60L, TimeUnit.SECONDS, // 空闲线程存活时间 new LinkedBlockingQueue<>(1000), // 工作队列 new ThreadFactoryBuilder().setNameFormat("io-pool-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略 ); // CPU密集型任务 - 使用较小的线程池 this.cpuBoundExecutor = new ThreadPoolExecutor( Runtime.getRuntime().availableProcessors(), // CPU核心数 Runtime.getRuntime().availableProcessors() * 2, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadFactoryBuilder().setNameFormat("cpu-pool-%d").build(), new ThreadPoolExecutor.AbortPolicy() ); } public CompletableFuture<String> processWithProperPool(String data) { return CompletableFuture.supplyAsync(() -> { // IO操作，使用IO线程池 return fetchFromDatabase(data); }, ioBoundExecutor); } public CompletableFuture<String> computeWithProperPool(String data) { return CompletableFuture.supplyAsync(() -> { // CPU密集型计算，使用CPU线程池 return heavyComputation(data); }, cpuBoundExecutor); } // 资源清理 @PreDestroy public void destroy() { ioBoundExecutor.shutdown(); cpuBoundExecutor.shutdown(); try { if (!ioBoundExecutor.awaitTermination(5, TimeUnit.SECONDS)) { ioBoundExecutor.shutdownNow(); } if (!cpuBoundExecutor.awaitTermination(5, TimeUnit.SECONDS)) { cpuBoundExecutor.shutdownNow(); } } catch (InterruptedException e) { ioBoundExecutor.shutdownNow(); cpuBoundExecutor.shutdownNow(); Thread.currentThread().interrupt(); } } }

线程池工作流程对比

三、异常为什么神秘消失了？

有些小伙伴在调试CompletableFuture时，经常会发现异常"神秘消失"了，这其实是CompletableFuture异常处理机制的一个特性。

异常丢失的典型案例

public class ExceptionDisappearance { public void testExceptionLost() { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { // 这里会抛出异常 return dangerousOperation(); }); // 添加转换链 CompletableFuture<String> resultFuture = future.thenApply(result -> { System.out.println("处理结果: " + result); return result + " processed"; }); try { // 这里不会抛出异常！ String result = resultFuture.get(); System.out.println("最终结果: " + result); } catch (Exception e) { // 异常被包装在ExecutionException中 System.out.println("捕获到异常: " + e.getClass().getName()); System.out.println("根本原因: " + e.getCause().getMessage()); } } private String dangerousOperation() { throw new RuntimeException("业务操作失败！"); } // 更隐蔽的异常丢失 public void testHiddenExceptionLoss() { CompletableFuture.supplyAsync(() -> { throw new BusinessException("重要异常"); }).thenAccept(result -> { // 如果上游有异常，这里不会执行 System.out.println("处理结果: " + result); }); // 程序继续执行，异常被忽略！ System.out.println("程序正常结束，但异常丢失了！"); } static class BusinessException extends RuntimeException { public BusinessException(String message) { super(message); } } }

CompletableFuture异常处理机制

正确的异常处理方式

public class ProperExceptionHandling { // 方法1：使用exceptionally进行恢复 public CompletableFuture<String> handleWithRecovery() { return CompletableFuture.supplyAsync(() -> { return riskyOperation(); }).exceptionally(throwable -> { // 异常恢复 System.err.println("操作失败，使用默认值: " + throwable.getMessage()); return "default-value"; }); } // 方法2：使用handle统一处理 public CompletableFuture<String> handleWithUnified() { return CompletableFuture.supplyAsync(() -> { return riskyOperation(); }).handle((result, throwable) -> { if (throwable != null) { // 处理异常 System.err.println("操作异常: " + throwable.getMessage()); return "error-value"; } return result + "-processed"; }); } // 方法3：使用whenComplete进行副作用处理 public CompletableFuture<Void> handleWithSideEffect() { return CompletableFuture.supplyAsync(() -> { return riskyOperation(); }).whenComplete((result, throwable) -> { if (throwable != null) { // 记录日志、发送告警等 logError(throwable); sendAlert(throwable); } else { // 正常业务处理 processResult(result); } }); } // 方法4：组合操作中的异常处理 public CompletableFuture<String> handleInComposition() { CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { return operation1(); }); CompletableFuture<String> future2 = future1.thenCompose(result1 -> { return CompletableFuture.supplyAsync(() -> { return operation2(result1); }); }); // 在整个链的末尾处理异常 return future2.exceptionally(throwable -> { Throwable rootCause = getRootCause(throwable); if (rootCause instanceof BusinessException) { return "business-fallback"; } else if (rootCause instanceof TimeoutException) { return "timeout-fallback"; } else { return "unknown-error"; } }); } private void logError(Throwable throwable) { // 记录错误日志 System.err.println("错误记录: " + throwable.getMessage()); } private void sendAlert(Throwable throwable) { // 发送告警 System.out.println("发送告警: " + throwable.getMessage()); } private Throwable getRootCause(Throwable throwable) { Throwable cause = throwable; while (cause.getCause() != null) { cause = cause.getCause(); } return cause; } }

四、回调地狱：当异步变成"异痛"

有些小伙伴在复杂业务场景中使用CompletableFuture时，很容易陷入回调地狱，代码变得难以理解和维护。

回调地狱的典型案例

public class CallbackHell { public CompletableFuture<String> processUserOrder(String userId) { return getUserInfo(userId) .thenCompose(userInfo -> { return getOrderHistory(userInfo.getId()) .thenCompose(orderHistory -> { return calculateDiscount(userInfo, orderHistory) .thenCompose(discount -> { return createOrder(userInfo, discount) .thenCompose(order -> { return sendConfirmation(userInfo, order); }); }); }); }); } // 上述代码的"平铺"版本，同样难以阅读 public CompletableFuture<String> processUserOrderFlat(String userId) { return getUserInfo(userId) .thenCompose(userInfo -> getOrderHistory(userInfo.getId())) .thenCompose(orderHistory -> getUserInfo(userId)) .thenCompose(userInfo -> calculateDiscount(userInfo, orderHistory)) .thenCompose(discount -> getUserInfo(userId)) .thenCompose(userInfo -> createOrder(userInfo, discount)) .thenCompose(order -> getUserInfo(userId)) .thenCompose(userInfo -> sendConfirmation(userInfo, order)); } }

结构化异步编程解决方案

public class StructuredAsyncProgramming { // 定义业务数据类 @Data @AllArgsConstructor public static class OrderContext { private String userId; private UserInfo userInfo; private List<Order> orderHistory; private Discount discount; private Order order; private String result; } public CompletableFuture<String> processUserOrderStructured(String userId) { OrderContext context = new OrderContext(userId, null, null, null, null, null); return getUserInfo(context.getUserId()) .thenCompose(userInfo -> { context.setUserInfo(userInfo); return getOrderHistory(userInfo.getId()); }) .thenCompose(orderHistory -> { context.setOrderHistory(orderHistory); return calculateDiscount(context.getUserInfo(), orderHistory); }) .thenCompose(discount -> { context.setDiscount(discount); return createOrder(context.getUserInfo(), discount); }) .thenCompose(order -> { context.setOrder(order); return sendConfirmation(context.getUserInfo(), order); }) .thenApply(result -> { context.setResult(result); return result; }) .exceptionally(throwable -> { // 统一异常处理 return handleOrderError(context, throwable); }); } // 使用thenCombine处理并行任务 public CompletableFuture<UserProfile> getUserProfile(String userId) { CompletableFuture<UserInfo> userInfoFuture = getUserInfo(userId); CompletableFuture<List<Order>> orderHistoryFuture = getOrderHistory(userId); CompletableFuture<List<Address>> addressesFuture = getUserAddresses(userId); return userInfoFuture.thenCombine(orderHistoryFuture, (userInfo, orders) -> { return new UserProfile(userInfo, orders, null); }).thenCombine(addressesFuture, (profile, addresses) -> { profile.setAddresses(addresses); return profile; }); } // 使用allOf处理多个独立任务 public CompletableFuture<Map<String, Object>> getDashboardData(String userId) { CompletableFuture<UserInfo> userInfoFuture = getUserInfo(userId); CompletableFuture<List<Order>> ordersFuture = getOrderHistory(userId); CompletableFuture<List<Notification>> notificationsFuture = getNotifications(userId); CompletableFuture<Preferences> preferencesFuture = getPreferences(userId); CompletableFuture<Void> allFutures = CompletableFuture.allOf( userInfoFuture, ordersFuture, notificationsFuture, preferencesFuture ); return allFutures.thenApply(v -> { Map<String, Object> dashboard = new HashMap<>(); try { dashboard.put("userInfo", userInfoFuture.get()); dashboard.put("orders", ordersFuture.get()); dashboard.put("notifications", notificationsFuture.get()); dashboard.put("preferences", preferencesFuture.get()); } catch (Exception e) { throw new CompletionException(e); } return dashboard; }); } }

异步编程模式对比

更推荐的方案：

五、内存泄漏：隐藏的资源消耗者

有些小伙伴可能没有意识到，不当使用CompletableFuture会导致内存泄漏，特别是在长时间运行的应用中。

内存泄漏的常见场景

public class MemoryLeakDemo { private final Map<String, CompletableFuture<String>> cache = new ConcurrentHashMap<>(); // 场景1：无限增长的缓存 public CompletableFuture<String> getDataWithLeak(String key) { return cache.computeIfAbsent(key, k -> { return CompletableFuture.supplyAsync(() -> fetchData(k)); }); } // 场景2：未完成的Future积累 public void processWithUnfinishedFutures() { for (int i = 0; i < 100000; i++) { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { // 模拟长时间运行或阻塞的任务 try { Thread.sleep(Long.MAX_VALUE); // 几乎永久阻塞 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "result"; }); // future永远不会完成，但一直存在于内存中 } } // 场景3：循环引用 public class TaskManager { private CompletableFuture<String> currentTask; private String status = "INIT"; public void startTask() { currentTask = CompletableFuture.supplyAsync(() -> { // 任务持有Manager的引用 while (!"COMPLETED".equals(status)) { // 处理任务 processTask(); } return "done"; }); } // Manager也持有任务的引用 public CompletableFuture<String> getCurrentTask() { return currentTask; } } }

内存泄漏检测和预防

public class MemoryLeakPrevention { private final Cache<String, CompletableFuture<String>> cache; public MemoryLeakPrevention() { // 使用Guava Cache自动清理 this.cache = CacheBuilder.newBuilder() .maximumSize(1000) .expireAfterAccess(10, TimeUnit.MINUTES) .removalListener((RemovalListener<String, CompletableFuture<String>>) notification -> { if (notification.getCause() == RemovalCause.SIZE || notification.getCause() == RemovalCause.EXPIRED) { // 取消未完成的任务 CompletableFuture<String> future = notification.getValue(); if (!future.isDone()) { future.cancel(true); } } }) .build(); } // 安全的缓存用法 public CompletableFuture<String> getDataSafely(String key) { try { return cache.get(key, () -> { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> fetchData(key)); // 添加超时控制 return future.orTimeout(30, TimeUnit.SECONDS) .exceptionally(throwable -> { // 发生异常时从缓存中移除 cache.invalidate(key); return "fallback-data"; }); }); } catch (ExecutionException e) { throw new RuntimeException(e); } } // 使用WeakReference避免循环引用 public static class SafeTaskManager { private WeakReference<CompletableFuture<String>> currentTaskRef; public void startTask() { CompletableFuture<String> task = CompletableFuture.supplyAsync(() -> { return performTask(); }); currentTaskRef = new WeakReference<>(task); // 任务完成后自动清理 task.whenComplete((result, error) -> { currentTaskRef = null; }); } } // 监控和诊断工具 public void monitorFutures() { // 定期检查未完成的Future Timer timer = new Timer(true); timer.scheduleAtFixedRate(new TimerTask() { @Override public void run() { int unfinishedCount = 0; for (CompletableFuture<?> future : cache.asMap().values()) { if (!future.isDone()) { unfinishedCount++; // 记录长时间运行的任务 if (future.isDoneExceptionally()) { // 处理异常任务 handleExceptionalFuture(future); } } } if (unfinishedCount > 100) { // 发出警告 System.err.println("警告: 有 " + unfinishedCount + " 个未完成的任务"); } } }, 0, 60000); // 每分钟检查一次 } private void handleExceptionalFuture(CompletableFuture<?> future) { // 处理异常Future，避免它们一直存在 future.exceptionally(throwable -> { // 记录异常日志 System.err.println("任务异常: " + throwable.getMessage()); return null; }); } }

内存泄漏检测流程

六、超时控制缺失

有些小伙伴在使用CompletableFuture时，经常会忘记设置超时控制，这可能导致线程永远阻塞。

超时问题的严重性

public class TimeoutPitfalls { // 危险的代码：没有超时控制 public String dangerousGet() { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { // 模拟网络问题导致的无限阻塞 return blockingNetworkCall(); }); try { // 如果任务永远不完成，这里会永远阻塞 return future.get(); } catch (Exception e) { return "error"; } } // 资源泄漏的示例 public void resourceLeakExample() { ExecutorService executor = Executors.newFixedThreadPool(10); for (int i = 0; i < 100; i++) { CompletableFuture.runAsync(() -> { try { // 长时间运行的任务 Thread.sleep(Long.MAX_VALUE); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, executor); } // 线程池中的线程都被占用，无法执行新任务 } private String blockingNetworkCall() { // 模拟网络问题 try { Thread.sleep(Long.MAX_VALUE); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "response"; } }

完整的超时控制方案

public class CompleteTimeoutSolution { private final ScheduledExecutorService timeoutExecutor; public CompleteTimeoutSolution() { this.timeoutExecutor = Executors.newScheduledThreadPool(2); } // 方法1：使用orTimeout（Java 9+） public CompletableFuture<String> withOrTimeout() { return CompletableFuture.supplyAsync(() -> { return externalServiceCall(); }).orTimeout(5, TimeUnit.SECONDS) // 5秒超时 .exceptionally(throwable -> { if (throwable instanceof TimeoutException) { return "timeout-fallback"; } return "error-fallback"; }); } // 方法2：使用completeOnTimeout（Java 9+） public CompletableFuture<String> withCompleteOnTimeout() { return CompletableFuture.supplyAsync(() -> { return externalServiceCall(); }).completeOnTimeout("timeout-default", 3, TimeUnit.SECONDS); } // 方法3：手动超时控制（Java 8兼容） public CompletableFuture<String> withManualTimeout() { CompletableFuture<String> taskFuture = CompletableFuture.supplyAsync(() -> { return externalServiceCall(); }); CompletableFuture<String> timeoutFuture = new CompletableFuture<>(); // 设置超时 timeoutExecutor.schedule(() -> { timeoutFuture.completeExceptionally(new TimeoutException("操作超时")); }, 5, TimeUnit.SECONDS); // 哪个先完成就返回哪个 return taskFuture.applyToEither(timeoutFuture, Function.identity()) .exceptionally(throwable -> { if (throwable instanceof TimeoutException) { return "manual-timeout-fallback"; } return "other-error-fallback"; }); } // 方法4：分层超时控制 public CompletableFuture<String> withLayeredTimeout() { return CompletableFuture.supplyAsync(() -> { return phase1Operation(); }).orTimeout(2, TimeUnit.SECONDS) .thenCompose(phase1Result -> { return CompletableFuture.supplyAsync(() -> { return phase2Operation(phase1Result); }).orTimeout(3, TimeUnit.SECONDS); }) .thenCompose(phase2Result -> { return CompletableFuture.supplyAsync(() -> { return phase3Operation(phase2Result); }).orTimeout(5, TimeUnit.SECONDS); }) .exceptionally(throwable -> { Throwable rootCause = getRootCause(throwable); if (rootCause instanceof TimeoutException) { // 根据超时阶段提供不同的降级策略 return "timeout-in-phase"; } return "general-fallback"; }); } // 方法5：可配置的超时策略 public CompletableFuture<String> withConfigurableTimeout(String operationType) { TimeoutConfig config = getTimeoutConfig(operationType); return CompletableFuture.supplyAsync(() -> { return performOperation(operationType); }).orTimeout(config.getTimeout(), config.getTimeUnit()) .exceptionally(throwable -> { return config.getFallbackStrategy().apply(throwable); }); } @PreDestroy public void destroy() { timeoutExecutor.shutdown(); try { if (!timeoutExecutor.awaitTermination(5, TimeUnit.SECONDS)) { timeoutExecutor.shutdownNow(); } } catch (InterruptedException e) { timeoutExecutor.shutdownNow(); Thread.currentThread().interrupt(); } } // 超时配置类 @Data public static class TimeoutConfig { private final long timeout; private final TimeUnit timeUnit; private final Function<Throwable, String> fallbackStrategy; } private TimeoutConfig getTimeoutConfig(String operationType) { switch (operationType) { case "fast": return new TimeoutConfig(1, TimeUnit.SECONDS, t -> "fast-timeout"); case "normal": return new TimeoutConfig(5, TimeUnit.SECONDS, t -> "normal-timeout"); case "slow": return new TimeoutConfig(30, TimeUnit.SECONDS, t -> "slow-timeout"); default: return new TimeoutConfig(10, TimeUnit.SECONDS, t -> "default-timeout"); } } }

超时控制策略

总结

通过上面的详细分析，我们可以看到CompletableFuture虽然强大，但也确实存在不少陷阱。

最后的建议

理解原理：不要只是机械地使用API，要理解CompletableFuture的工作原理
适度使用：不是所有场景都需要异步，同步代码更简单易懂
测试覆盖：异步代码的测试很重要，要覆盖各种边界情况
监控告警：在生产环境中要有完善的监控和告警机制
持续学习：关注Java并发编程的新特性和最佳实践

记住，工具是为了提高生产力，而不是制造问题。

掌握了这些避坑技巧，CompletableFuture将成为你手中强大的并发编程利器！

文章转载自：苏三说技术
原文链接：https://www.cnblogs.com/12lisu/p/19216660
体验地址：http://www.jnpfsoft.com/?from=001YH

CompletableFuture的5大坑！

前言