CompletableFuture 详解

CompletableFuture 介绍

1. 基本概念

CompletableFuture 是 Java 8 引入的异步编程工具，用于简化非阻塞式任务编排。核心特点：

• 支持链式调用和组合操作
• 提供异常处理机制
• 可自定义线程池

2. 结果获取与触发计算

CompletableFuture<String> future = new CompletableFuture<>();
future.complete("Result"); // 手动触发计算结果
future.get(); // 阻塞获取结果
future.join(); // 同get但不抛受检异常
future.completeExceptionally(new RuntimeException()); // 显式设置异常

3. 结果处理

• 转换结果 (thenApply):

  future.thenApply(s -> s.toUpperCase())
        .thenAccept(System.out::println); // 输出大写结果
  

• 异常恢复 (exceptionally):

  future.exceptionally(ex -> "Fallback Value");
  

• 结果处理 (handle):

  future.handle((res, ex) -> ex != null ? "Error" : res);
  

4. 结果消费

• 无返回值消费 (thenAccept):

  future.thenAccept(r -> System.out.println("Received: " + r));
  

• 最终操作 (thenRun):

  future.thenRun(() -> System.out.println("Processing complete"));
  

5. 线程池选择

默认使用 ForkJoinPool.commonPool()，可通过参数指定线程池：

ExecutorService customPool = Executors.newFixedThreadPool(10);
future.thenApplyAsync(s -> process(s), customPool); // 使用自定义线程池

最佳实践：

• CPU密集型任务：使用固定大小线程池（核数+1）
• IO密集型任务：使用缓存线程池

6. 计算速度优化

• 竞速模式 (anyOf):

  CompletableFuture.anyOf(future1, future2)
                   .thenAccept(firstResult -> useFastest(firstResult));
  

• 超时控制 (orTimeout):

  future.orTimeout(2, TimeUnit.SECONDS) // JDK9+
        .exceptionally(ex -> "Timeout Fallback");
  

7. 结果合并

• 二元合并 (thenCombine):

  future1.thenCombine(future2, (res1, res2) -> res1 + res2);
  

• 多元合并 (allOf):

  CompletableFuture.allOf(futures)
                  .thenApply(v -> Arrays.stream(futures)
                                       .map(CompletableFuture::join)
                                       .collect(Collectors.toList()));
  

大厂实践案例：电商订单处理（阿里双11场景）

场景需求

用户下单后需并行执行：

1. 库存校验
2. 风控审核
3. 优惠计算
4. 物流预检

实现方案

// 1. 定义子任务
CompletableFuture<Boolean> stockCheck = supplyAsync(() -> checkStock(order), ioPool);
CompletableFuture<RiskResult> riskCheck = supplyAsync(() -> riskControl(order), ioPool);
CompletableFuture<Coupon> couponCalc = supplyAsync(() -> calcCoupon(order), cpuPool);
CompletableFuture<Logistics> logisticsPrep = supplyAsync(() -> prepareLogistics(order), ioPool);

// 2. 合并结果
CompletableFuture.allOf(stockCheck, riskCheck, couponCalc, logisticsPrep)
                .thenApply(v -> {
                    if (!stockCheck.join()) throw new StockException();
                    return new OrderResult(
                        riskCheck.join(),
                        couponCalc.join(),
                        logisticsPrep.join()
                    );
                })
                .exceptionally(ex -> {
                    monitor.logError(ex);  // 异常监控
                    return fallbackHandler(order); // 降级处理
                })
                .thenAccept(this::sendNotification); // 结果通知

性能优化策略

1. 线程池隔离：
   • IO任务（网络调用）使用 CachedThreadPool
   • CPU计算使用 FixedThreadPool
2. 超时熔断：

   riskCheck.orTimeout(500, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> DEFAULT_RISK_RESULT);
   

3. 优先级调度：

   stockCheck.thenRun(() -> logisticsPrep.cancel(false)); // 库存失败时取消物流预检
   

收益对比

注：某电商平台2022年双11实战数据，订单峰值58.3万笔/秒

通过合理使用 CompletableFuture 的异步组合能力，可显著提升系统吞吐量和响应速度，尤其适用于高并发微服务场景。