CompletableFuture异步编程

一、核心定位

CompletableFuture = JDK8 全新异步非阻塞编程工具

• 替代 Future + 线程池 老旧写法
• 支持链式回调、任务串行 / 并行组合、异常自动传递
• 不用手动写回调嵌套，解决回调地狱
• 默认用 ForkJoinPool 公共线程池，也可自定义指定线程池

---

二、异步任务创建与执行

1. 常用创建方法

// 无返回值 异步执行
CompletableFuture.runAsync(Runnable)

// 有返回值 异步执行
CompletableFuture.supplyAsync(Supplier)

2. 默认线程池规则

• 不传线程池：默认使用ForkJoinPool.commonPool()公共池
• 生产不推荐默认池：全局共用、容易互相影响、OOM / 阻塞传染
• 最佳实践：所有异步手动传入自定义线程池

示例：

// 有返回值 + 指定业务线程池
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "异步结果";
}, businessPool);

---

三、链式回调三大核心方法

1. thenApply：有入参、有返回

接收上一步结果，处理后返回新结果，串行流转

supplyAsync(() -> 100)
.thenApply(num -> num * 2)
.thenApply(res -> res + "ok");

2. thenAccept：有入参、无返回

消费上一步结果，只做业务处理，无返回值

.thenAccept(result -> {
    System.out.println("拿到结果：" + result);
});

3. thenCompose：任务嵌套、扁平化串行

用于把一个 CompletableFuture 结果，再开启下一个异步 Future

解决多层 Future 嵌套

// 场景：异步查用户 -> 异步查订单
CompletableFuture<User> userFuture = getUser();
CompletableFuture<Order> orderFuture = 
    userFuture.thenCompose(user -> getOrderAsync(user.getId()));

thenRun() / thenRunAsync()

• 不需要上一步的返回值
• 自身也没有入参
• 只作用：上一个异步任务执行完毕，再执行一段后置逻辑
• 无参、无结果消费，只做「收尾、通知、打印、埋点」

import java.util.concurrent.CompletableFuture;

public class ThenRunDemo {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            // 第一步业务
            System.out.println("第一步：异步查询数据");
            return "订单数据";
        })
        .thenRun(() -> {
            // 不接收上面的返回值，纯后置动作
            System.out.println("thenRun 执行：任务完成，打印日志/埋点/发送通知");
        })
        .join();
    }
}

---

四、多任务组合（并行调度核心）

1. allOf：所有任务全部完成才继续

全部成功 / 全部结束 才触发后续，适合批量并行处理

CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);
all.join(); // 等待全部执行完毕

2. anyOf：任意一个任务完成就继续

只要其中一个执行完成，立刻往下走

适合：多接口兜底、多数据源择优、超时快速响应

CompletableFuture<Object> any = CompletableFuture.anyOf(f1, f2);

---

五、异常处理

1. exceptionally：异常兜底返回默认值

只捕获异常，正常流程不执行

.exceptionally(ex -> {
    log.error("异步任务异常", ex);
    return "兜底默认值";
});

2. handle：正常 + 异常都能捕获

无论成功 / 失败都会执行，最通用

.handle((res, ex) -> {
    if (ex != null) {
        // 异常处理
    }
    return res;
});

3. whenComplete

接收结果或异常，无返回，只做收尾、日志、清理

.whenComplete((res, ex) -> {
    // 后置收尾逻辑
});

---

六、线程切换 & 线程池指定

1. 不带 Async 方法

thenApply / thenAccept

• 共用上一步的线程执行回调
• 同一线程串行跑完

2. 带 Async 方法

thenApplyAsync / thenAcceptAsync

• 重新抢占线程池线程
• 可以单独指定第二个线程池，拆分不同业务线程隔离

示例：

// 第一段：A线程池
supplyAsync(() -> queryData(), poolA)
// 第二段：切换到 B线程池 异步回调
.thenApplyAsync(data -> convert(data), poolB);

关键开发规范

1. 禁止直接用默认 commonPool
2. 不同业务、不同耗时（IO / 计算）拆分独立线程池
3. 异步链路全链路指定线程池，避免公共池阻塞雪崩

---

七、常用阻塞等待方法

• get()：阻塞等待，抛受检异常
• join()：阻塞等待，只抛运行时异常，开发常用
• get(timeout, unit)：限时等待，防止永久阻塞

---

八、极简核心总结

1. 创建
   • runAsync 无返回；supplyAsync 有返回
   • 生产必须手动传入自定义线程池
2. 串行回调
   • thenApply：加工 + 返回
   • thenAccept：消费无返回
   • thenCompose：扁平化嵌套异步
3. 多任务组合
   • allOf：全部完成
   • anyOf：任意一个完成
4. 异常
   • exceptionally：异常兜底
   • handle：成功异常全覆盖
   • whenComplete：后置收尾
5. 线程模型
   • 无 Async：复用原有线程
   • 有 Async：重新从线程池拿线程，支持多池隔离