CompletableFuture异步编程

ℹ️本篇定位

线程池能"并行"，但多个异步任务之间有依赖、要合并、要异常兜底时，用 Future+get() 手写编排是灾难。CompletableFuture（JDK 8）把异步任务变成一条声明式流水线——链式串联、并行合并、任选其一、异常兜底，全都能写出来。本篇前置：03-线程池与 07-Lambda与函数式接口。

一、问题提出：老 Future 的三个痛点

ExecutorService pool = Executors.newFixedThreadPool(4);
Future<Order> f1 = pool.submit(() -> queryOrder(id));
Future<User> f2 = pool.submit(() -> queryUser(uid));
Order order = f1.get();   // 1. get() 阻塞主线程，等的时间不能干别的
User user = f2.get();     // 2. 多个 Future 只能逐个 get，无法"两个都好了再做"
// 3. 任务之间要串行（拿到订单再查用户），只能自己 sleep + 循环轮询

CompletableFuture 提供 ~Async 系列方法，把"等"也交给框架。

二、创建：四把钥匙

 // 1. 无返回值的动作
CompletableFuture<Void> f = CompletableFuture.runAsync(() -> sendMsg(user));
 // 2. 有返回值
CompletableFuture<Order> f2 = CompletableFuture.supplyAsync(() -> queryOrder(id));
 // 3. 指定线程池（生产必加，别用默认的 ForkJoin 公共池）
CompletableFuture.supplyAsync(() -> queryUser(uid), bizPool);
 // 4. 手动完成（回调/外部系统触发的适配层）
CompletableFuture<String> manual = new CompletableFuture<>();
manual.complete("done");        // 手动塞结果
manual.completeExceptionally(e); // 手动塞异常

默认线程池要警惕：不加参数的 runAsync/supplyAsync 跑在 ForkJoinPool.commonPool() 上，它是全局共享、并行度 = CPU 核数-1——IO 密集任务必须显式传自己的线程池，否则可能拖垮整个 JVM 的公共池。

三、编排：单线串行（then*）

supplyAsync(() -> queryOrder(id))                          // 查订单
   .thenApply(order -> order.getAmount())                  // 同步转换（同一线程）
   .thenApplyAsync(amount -> calcTax(amount), pool)        // 异步转换（换线程）
   .thenAccept(tax -> System.out.println(tax))             // 消费结果，不返回
   .thenRun(() -> System.out.println("done"));             // 只要前一步完成就执行

方法名口诀：thenApply=变换、thenAccept=消费、thenRun=只跑动作；带 Async 后缀 = 换线程执行，可指定线程池。

四、编排：多线合并与竞赛

// AND：两个都完成再做（合并两路结果）
CompletableFuture<String> userName = supplyAsync(() -> queryUser(uid), pool);
CompletableFuture<Order> order = supplyAsync(() -> queryOrder(id), pool);
userName.thenCombine(order,
        (u, o) -> String.format("%s 的订单 %s", u, o.getId()))
       .thenAccept(System.out::println);

// ALL / ANY：等全部 / 等任意一个
CompletableFuture.allOf(userName, order).join();          // 全部完成
CompletableFuture.anyOf(userName, order).join();          // 任一完成即返回（返回 Object）

// 竞速：两个路径谁先到用谁
supplyAsync(this::dbQuery).applyToEither(
    supplyAsync(this::cacheQuery), r -> r);

五、异常兜底：异步世界里的 try-catch

supplyAsync(() -> queryOrder(id))
    .exceptionally(ex -> { log.error("查询失败,回退", ex); return Order.DEFAULT; })   // 异常→默认值
    .handle((order, ex) -> ex != null ? fallback(order) : enrich(order));              // 正常和异常都处理
    .whenComplete((order, ex) -> {                           // 无论成败的收尾动作
        if (ex != null) log.error("失败", ex);
        else metrics.record(order);
    });

锅在哪烧就在哪接：exceptionally/handle 只处理自己之前那一段的异常，不能隔好几个 then 再统一 catch——每个可能出错的环节就近兜底。

六、三大易错点

get()/join() 阻塞：get() 抛受检异常（InterruptedException），join() 抛非受检（CompletionException）——两者都会阻塞线程，链式写法里尽量用回调，最后才 join 一次收口；
默认线程池：见第二节，IO 任务必须显式指定业务线程池；
thenApply vs thenApplyAsync 的坑：thenApply 在上一步完成线程上同步执行，任务很重会占住上一个工作线程——重的转换用 thenApplyAsync 换到线程池。判断标准：当前步骤是否耗 CPU / 是否想换线程。

七、高频面试题速答

CompletableFuture 和 Future 区别？ Future 只能 get 阻塞等结果；CompletableFuture 支持链式回调（then*）、多路合并（allOf/anyOf/combine）、异常兜底（exceptionally/handle）。
thenApply 和 thenAccept 区别？ thenApply 返回新结果（变换），thenAccept 只消费不返回。
默认线程池是什么？为什么不能用来跑 IO？ ForkJoinPool.commonPool，全局共享、并行度 CPU 核数-1，IO 任务占满会饿死其他使用方——显式传线程池。
allOf 和 anyOf 返回值是什么？ allOf 返回 CompletableFuture<Void>（需要再 thenApply 取各结果），anyOf 返回第一个完成的结果（Object，要强转）。真正并行的底层是 03-线程池。

勾连

线程池参数与拒绝策略：03-线程池
底层 Future/FutureTask 与 Callable：01-线程基础与生命周期
编排思想的流式镜像：08-Stream与Optional

⬅️ 05-并发容器与工具类 🏠 00-Java ➡️ 07-并发高频面试题

06-CompletableFuture异步编程