`CompletableFuture` 异步编排源码精读

概述

CompletableFuture 把异步结果（Future）与回调编排（CompletionStage）合二为一：不阻塞等待，而是声明"结果到来后做什么"，由框架调度执行。本文基于 OpenJDK 21 源码拆解其内部状态、依赖推拉机制与组合实现。

一、内部状态与完成机制

1.1 核心字段

java

// java.util.concurrent.CompletableFuture
volatile Object result;            // 完成结果
volatile Completion stack;         // 依赖动作栈（链表，头插）

// result 的三种形态：
//   null            → 未完成
//   T               → 正常结果（直接存值）
//   AltResult       → 异常结果（包装 Throwable）
static final class AltResult {
    final Throwable ex;   // 为 null 表示取消
}

依赖动作用单向链表栈（stack 字段，头插法）存储，多个 thenApply 会叠加成栈。完成时遍历栈逐个触发，无需锁——靠 CAS 保证线程安全。

1.2 `complete(T)` 的 CAS 设置

java

public boolean complete(T value) {
    boolean triggered = completeValue(value);
    // 完成成功后，postComplete() 触发所有依赖动作
    postComplete();
    return triggered;
}

private boolean completeValue(T t) {
    return U.compareAndSetReference(this, RESULT, null, t);   // CAS null → 结果
}

CAS 保证"只完成一次"：多个线程并发 complete 只有一个成功返回 true，其余返回 false。这是并发编程中 complete 竞态的核心保障。

1.3 `postComplete` 触发依赖

java

final void postComplete() {
    CompletableFuture<?> f = this; Completion h;
    while ((h = f.stack) != null) {
        f.stack = h.next;              // 弹出栈顶
        if (h.tryFire(NESTED)) {       // 触发依赖动作
            // 动作自己完成了一个新 future → 继续触发它的依赖
            f = h.dep;                 // 沿依赖链往下走
        }
    }
}

依赖触发是级联的：A.thenApply(fn) 生成新 future B，A 完成时触发 B 的 tryFire，B 若也因此完成，继续触发 B 的依赖……直到链尾。

二、异步执行：`supplyAsync` 与 `ForkJoinPool`

java

// CompletableFuture
private static final Executor ASYNC_POOL = ForkJoinPool.commonPool();

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(ASYNC_POOL, supplier);
}

static <U> CompletableFuture<U> asyncSupplyStage(Executor e, Supplier<U> f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<U> d = new CompletableFuture<>();
    e.execute(new AsyncSupply<U>(d, f));   // 提交到 ForkJoinPool.commonPool
    return d;
}

java

// AsyncSupply 动作
static final class AsyncSupply<T> extends ForkJoinTask<Void>
        implements Runnable, AsynchronousCompletionTask {
    final CompletableFuture<T> dep;   // 目标 future
    final Supplier<T> fn;
    public void compute() { run(); }  // ForkJoinTask 适配
    public void run() {
        CompletableFuture<T> d; Supplier<T> f;
        if ((d = dep) != null && (f = fn) != null) {
            dep = null; fn = null;
            if (d.result == null) {
                try {
                    d.completeValue(f.get());      // 执行 Supplier 并完成
                } catch (Throwable ex) {
                    d.completeThrowable(ex);       // 异常 → AltResult
                }
            }
            d.postComplete();                      // 触发后续依赖
        }
    }
}

ForkJoinPool.commonPool() 是全局共享池（大小 = 核数 - 1），线程为守护线程。未指定 Executor 的异步方法默认用它。

三、依赖编排：`thenApply` 与 `UniApply`

3.1 `thenApply(Function)`

java

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) {
    return uniApplyStage(null, fn);    // 同步模式（e == null）
}

public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) {
    return uniApplyStage(ASYNC_POOL, fn);   // 异步模式
}

private <U> CompletableFuture<U> uniApplyStage(Executor e, Function<? super T,? extends U> f) {
    CompletableFuture<U> d = new CompletableFuture<U>();  // 新 future（下游）
    if (f != null && d.uniApply(this, f, e == null ? null : new UniApply<T,U>(e, f, d, this))) {
        return d;   // 上游已完成后立即执行
    }
    // 上游未完成 → 把动作压入上游的 stack
    push(new UniApply<T,U>(e, f, d, this));
    return d;
}

java

static final class UniApply<T,U> extends UniCompletion<T,U> {
    final Function<? super T,? extends U> fn;
    final boolean tryFire(int mode) {
        CompletableFuture<U> d; CompletableFuture<T> a;
        if ((d = dep) == null || !d.uniApply(a = src, fn, mode > 0 ? null : this))
            return false;
        dep = null; src = null; fn = null;   // 触发后清空，帮助 GC
        return true;
    }
}

final <S> boolean uniApply(CompletableFuture<S> a, Function<? super S,? extends T> f, UniApply<S,T> node) {
    Object r;
    if ((r = a.result) == null) return false;       // 上游未完成 → 挂起
    if (f == null) {
        if (result == null)
            result = r instanceof AltResult ? r : new AltResult(new NullPointerException());
    } else {
        try {
            T t = f.apply((S)(r instanceof AltResult ? null : r));  // 执行函数
            completeValue(t);                        // 完成下游
        } catch (Throwable ex) {
            completeThrowable(ex);                   // 异常传播
        }
    }
    return true;
}

流程总结：

thenApply 的两种路径：
  ① 上游已完成后调用 → uniApply 立即执行函数，同步返回结果
  ② 上游未完成 → 创建 UniApply 节点 push 到上游 stack
     上游 complete 时 postComplete 遍历栈 → tryFire → uniApply

3.2 `thenCompose` 的扁平化

thenCompose 的 Function 返回一个 future，需要"展开"而非嵌套：

java

public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
    return uniComposeStage(null, fn);
}

// UniCompose.tryFire 核心：
// ① 上游完成后，取 fn 结果（一个新 future）
// ② 把下游 d 的依赖挂到新 future 上 → 结果扁平化，不会出现 future<future>

对比：thenApply(x -> future) 得到 CompletableFuture<CompletableFuture<U>>（嵌套）；thenCompose 自动拍平为 CompletableFuture<U>。

四、组合：`allOf` / `anyOf`

4.1 `allOf` — 全部完成

java

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
    return andTree(cfs, 0, cfs.length - 1);   // 分治成两两 AndRelay 树
}

// 简化逻辑：为每个 future 挂一个计数完成动作
// 记录剩余待完成数 n，每个 future 完成时 n-1，减到 0 才 complete 汇总节点

每个入参 future 完成时触发 AndRelay，内部维护剩余计数；只有所有入参都完成后，汇总 future 才完成（结果值为 null）。分治树把 O(n) 依赖压成 O(log n) 深度。

4.2 `anyOf` — 任一完成

java

public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
    // 为每个入参挂 OrRelay，任一完成即 complete 汇总节点
}

任一入参先完成（成功或异常），OrRelay 立即把该结果透传给汇总 future，其余入参的后续完成被忽略。

五、超时控制

5.1 `orTimeout(long, TimeUnit)`

java

public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) {
    if (unit == null) throw new NullPointerException();
    if (result == null)
        // 注册一个延迟完成动作：到时若仍未完成 → 补一个 TimeoutException
        whenComplete(new Canceller(Delayer.delay(
            new Timeout(this), timeout, unit)));
    return this;
}

static final class Timeout implements Runnable {
    final CompletableFuture<?> f;
    public void run() {
        if (f != null && !f.isDone())
            f.completeExceptionally(new TimeoutException());   // 超时异常完成
    }
}

Delayer 内部用 ScheduledThreadPoolExecutor 延迟调度，到时若 future 仍未完成，注入 TimeoutException。

5.2 `completeOnTimeout(T, long, TimeUnit)`

java

public CompletableFuture<T> completeOnTimeout(T value, long timeout, TimeUnit unit) {
    if (unit == null) throw new NullPointerException();
    if (result == null)
        whenComplete(new Canceller(Delayer.delay(
            new DelayedCompleter<T>(this, value), timeout, unit)));
    return this;
}

// 到时未完成 → 用默认值完成（不抛异常）
static final class DelayedCompleter<U> implements Runnable {
    final CompletableFuture<U> f; final U u;
    public void run() {
        if (f != null && !f.isDone()) f.complete(u);
    }
}

方法	超时行为
`orTimeout`	超时抛 `TimeoutException`（调用方感知失败）
`completeOnTimeout`	超时返回默认值（容忍延迟，正常继续）
`get(timeout)`	同步阻塞等超时，抛 `TimeoutException`（调用线程等待）

六、异常处理

java

public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) {
    return uniExceptionallyStage(fn);
}

final <T> boolean uniExceptionally(CompletableFuture<T> a,
                                   Function<? super Throwable, ? extends T> fn) {
    Object r;
    if ((r = a.result) == null) return false;    // 上游未完成
    if (result == null) {
        try {
            if (r instanceof AltResult) {
                // 上游是异常 → 执行 fn，把异常换成恢复值
                completeValue(fn.apply(((AltResult) r).ex));
            } else {
                // 上游正常 → 直接透传结果（不执行 fn）
                completeValue((T) r);
            }
        } catch (Throwable ex) {
            completeThrowable(ex);
        }
    }
    return true;
}

exceptionally 语义：
  上游异常 → fn 执行，返回恢复值（异常被消化）
  上游正常 → fn 不执行，结果直接透传

同族的 handle（无论成败都执行，参数为 (value, throwable)）、whenComplete（无论成败都执行但透传原结果/异常）属于不同侧重点：

thenApply     成功才执行，返回新值
thenCompose   成功才执行，返回扁平化 future
exceptionally 异常才执行，返回恢复值
handle        成败都执行，可返回新值（吞异常）
whenComplete  成败都执行，透传原结果/异常

七、实现要点

CompletableFuture 核心：
  内部状态：volatile result（null/值/AltResult 异常）+ Completion 依赖栈
  完成机制：CAS null → 结果，postComplete 级联触发依赖
  异步执行：ForkJoinPool.commonPool 共享池（AsyncSupply）
  thenApply：UniApply 节点，完成即执行 / 未完成则压栈
  thenCompose：UniCompose 扁平化，避免 future<future>
  allOf：AndRelay 计数全完成（分治树）
  anyOf：OrRelay 任一完成
  超时：Delayer（ScheduledThreadPoolExecutor）注入 TimeoutException / 默认值
  异常：exceptionally 异常时执行 fn 恢复

常见陷阱：
  回调中做耗时操作 → 阻塞 ForkJoinPool 公共池（指定 Executor）
  忘记异常处理 → 异常被静默吞掉
  thenApply 返回 future → 应该用 thenCompose
  allOf 参数为空数组 → 立即完成
  get() 无限阻塞 → 用 orTimeout 或 get(timeout)

CompletableFuture 异步编排源码精读 ​

概述 ​

一、内部状态与完成机制 ​

1.1 核心字段 ​

1.2 complete(T) 的 CAS 设置 ​

1.3 postComplete 触发依赖 ​

二、异步执行：supplyAsync 与 ForkJoinPool ​

三、依赖编排：thenApply 与 UniApply ​

3.1 thenApply(Function) ​

3.2 thenCompose 的扁平化 ​

四、组合：allOf / anyOf ​

4.1 allOf — 全部完成 ​

4.2 anyOf — 任一完成 ​

五、超时控制 ​

5.1 orTimeout(long, TimeUnit) ​

5.2 completeOnTimeout(T, long, TimeUnit) ​

六、异常处理 ​

七、实现要点 ​

`CompletableFuture` 异步编排源码精读

概述

一、内部状态与完成机制

1.1 核心字段

1.2 `complete(T)` 的 CAS 设置

1.3 `postComplete` 触发依赖

二、异步执行：`supplyAsync` 与 `ForkJoinPool`

三、依赖编排：`thenApply` 与 `UniApply`

3.1 `thenApply(Function)`

3.2 `thenCompose` 的扁平化

四、组合：`allOf` / `anyOf`

4.1 `allOf` — 全部完成

4.2 `anyOf` — 任一完成

五、超时控制

5.1 `orTimeout(long, TimeUnit)`

5.2 `completeOnTimeout(T, long, TimeUnit)`

六、异常处理

七、实现要点