books

《Java 并发编程·JUC 源码与容器》13.1 CompletableFuture 的依赖栈

thenApply 返回的 future 没有线程在等它，为什么 complete 一调用整条链就醒了？本节从 result 与 stack 两个 volatile 字段入手，拆解 Treiber 栈的压栈与倒栈、NESTED 模式的迭代展开，并用反射读出真实链上的 UniApply、UniCompose、UniRelay、CoCompletion。

Leeting Yan 2026-10-28 11 分钟阅读 5411 字

承接的问题

卷一第 13 章讲的是怎么用：thenApply 串行、thenCompose 展平、handle 收尾。用的时候有一件事说不通——f0.thenApply(...) 返回的 future 并没有任何线程在等它，可 f0.complete("x") 一调用，整条链就在调用 complete 的那条线程上跑完了，一条线程都没新建。

本卷不重复用法，只回答三个实现问题：一个 future 用什么结构记住"等着我的动作"？complete 之后这些动作为什么能在调用者的线程上跑？链条很长时为什么不会栈溢出？

答案集中在两个字段上：result 表示"我完成了没有"，stack 表示"谁在等我"。CompletableFuture 本质上是一个"结果 + 依赖栈"的对象，回调不是注册在某个调度器上，而是挂在这个对象自己的栈里。

result 与 stack

这两个字段在源码里挨着声明：

// JDK 25 源码摘录，已删减
volatile Object result;       // Either the result or boxed AltResult
volatile Completion stack;    // Top of Treiber stack of dependent actions

final boolean internalComplete(Object r) { // CAS from null to r
    return RESULT.compareAndSet(this, null, r);
}

final boolean tryPushStack(Completion c) {
    Completion h = stack;
    NEXT.set(c, h);         // CAS piggyback
    return STACK.compareAndSet(this, h, c);
}

result 一个字段承担四种状态：

取值	含义
`null`	未完成
普通对象	正常完成，值就是它
`NIL`（`AltResult` 单例，`ex == null`）	正常完成，值是 `null`
`AltResult`（`ex != null`）	异常完成或已取消

“完成"因此就是一次 RESULT.compareAndSet(this, null, r)，只有第一次成功。没有单独的 done 标志，也不需要先读状态再写值——状态和值在同一个字里，天然原子。13.3 会讲为什么异常也要包成一个"结果”。

stack 是 Treiber 栈顶，压栈三步：读栈顶 h、把 c.next 指向 h、CAS 把栈顶从 h 换成 c。NEXT.set(c, h) 用 set 而不是 CAS，因为此刻 c 还是线程私有的、没被任何人看见；等 STACK 的 CAS 成功，c 连同它的 next 才一起对别人可见。源码注释把这叫 “CAS piggyback”——一次 CAS 顺带发布了整个节点。

压栈失败意味着什么

真正的巧处在 unipush，它处理"注册"和"完成"两个动作撞车的窗口：

// JDK 25 源码摘录，已删减
final void unipush(Completion c) {
    if (c != null) {
        while (!tryPushStack(c)) {
            if (result != null) {
                NEXT.set(c, null);
                break;
            }
        }
        if (result != null)
            c.tryFire(SYNC);
    }
}

压栈失败只有一种原因——栈顶被别的线程改了；而无论压栈成功后再读一次 result，还是压栈失败后读 result，只要读到非 null，就说明完成者可能已经错过了我，于是自己就地触发。

这是一次两边都"先写自己的标志、再读对方的标志"的握手：

完成者：RESULT.compareAndSet(null, r) 写 result，然后 postComplete() 读 stack。
注册者：CAS 改 stack，然后读 result。

两个 volatile 读写交叉，至少有一方能看到对方：完成者若在注册者压栈后才读 stack，就会弹到新节点；若在之前读，那么注册者的 result 读必然发生在写之后，会看到非 null 并自己 tryFire。没有锁，也不会丢唤醒，前提是两侧都遵守这个顺序。## 21 个 Completion 子类

栈上的节点不是同一种东西。Completion 继承 ForkJoinTask<Void> 并实现 Runnable、AsynchronousCompletionTask，把"待执行的回调"伪装成 ForkJoin 任务，这样它能被丢进任何池子里跑，claim() 还能借用 ForkJoinTask 的 tag 位做"只允许一个线程认领"。

子类	对应 API	触发条件
`UniApply` / `UniAccept` / `UniRun`	`thenApply` / `thenAccept` / `thenRun`	单源完成
`UniWhenComplete` / `UniHandle`	`whenComplete` / `handle`	单源完成，正常或异常都触发
`UniExceptionally` / `UniComposeExceptionally`	`exceptionally` / `exceptionallyCompose`	单源异常完成
`UniCompose` / `UniRelay`	`thenCompose` / `copy`	单源完成；`UniRelay` 只搬结果不跑用户函数
`BiApply` / `BiAccept` / `BiRun` / `BiRelay`	`thenCombine` 四兄弟 / `allOf`	两个源都完成
`CoCompletion`	上面四个的"另一半"	包的还是同一个 `BiCompletion`
`OrApply` / `OrAccept` / `OrRun`	`applyToEither` 系列	任一源完成
`AnyOf` / `Signaller`	`anyOf` / 阻塞等待	任一完成 / 唤醒等待线程

继承链只有两层：UniCompletion<T,V> 带 src、dep、executor；BiCompletion 在它上面加一个 snd。一个回调节点同时记住"我依赖谁"（src）和"我完成谁"（dep），所以依赖图不需要中心化注册表，图就长在对象自己身上。

postComplete：把栈倒出来逐个触发

complete 的最后一步永远是 postComplete()：

// JDK 25 源码摘录，已删减
final void postComplete() {
    CompletableFuture<?> f = this; Completion h;
    while ((h = f.stack) != null ||
           (f != this && (h = (f = this).stack) != null)) {
        CompletableFuture<?> d; Completion t;
        if (STACK.compareAndSet(f, h, t = h.next)) {
            if (t != null) {
                if (f != this) {
                    pushStack(h);
                    continue;
                }
                NEXT.compareAndSet(h, t, null); // try to detach
            }
            f = (d = h.tryFire(NESTED)) == null ? this : d;
        }
    }
}

这段循环是整节最值得抄在纸上的十行：

f 是"当前正在倒的栈"，不是固定的 this。 起初 f == this；某个回调触发后把它的 dep（下游 future）返回出来，f 就切到下游继续倒。一条 A → B → C 的链，是同一条循环依次处理三个栈，而不是三层函数调用。
分叉被推回根栈。 如果 f != this 且弹出的节点下面还有别的节点（t != null），说明这个栈上不止一个依赖，当前只沿一条路径走，于是把 h 压回根 future 的栈再 continue。源码注释里的 “pushing others to avoid unbounded recursion” 就是这一句。
tryFire 返回 null 表示"这条路径到此为止"，f 回到 this；返回一个 future 表示"它自己也有依赖要触发"，顺着它继续。
弹栈用 CAS，因为可能和新的 thenApply 注册并发；NEXT.compareAndSet(h, t, null) 只是"尝试"断开链接，失败也无所谓。

栈是后进先出，所以同一个 future 上注册多个回调，触发顺序与注册顺序相反。本机输出：

触发顺序 = [third-registered, second-registered, first-registered]

（在同一个未完成的 future 上依次挂三个 thenApply，各自往列表里追加名字，再 complete。）这不是实现瑕疵，而是"无锁栈 + 就地触发"的必然结果。任何依赖"先注册先执行"的代码都不该建立在 CF 上。

NESTED 模式与递归展开

postComplete 调用的是 tryFire(NESTED)，而 Completion.run() 调用的是 tryFire(ASYNC)。三个模式的区别不在"跑不跑"，而在"谁来跑、跑完怎么交还"：

模式	值	调用点	语义
`SYNC`	0	`unipush` / `bipush` / `orpush` 发现源已完成	就地跑，但必须先 `claim()`
`ASYNC`	1	`Completion.run()` / `exec()`	已在执行器线程上，跳过 `claim()`
`NESTED`	-1	`postComplete()`	就地跑并 `claim()`，跑完把依赖交回外层循环

claim() 是"认领"：

// JDK 25 源码摘录，已删减
final boolean claim() {
    Executor e = executor;
    if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {
        if (e == null)
            return true;
        executor = null; // disable
        e.execute(this);
    }
    return false;
}

一次 tag 位 CAS 保证同一个 Completion 只被执行一次。返回 true 表示"归我了，我这就跑"；返回 false 表示"别人拿走了，或者我已经把它交给了执行器"。*Async 变体就是这么工作的：模式为 SYNC/NESTED 时 claim() 把节点 execute 出去并返回 false，等执行器线程调用 run() → tryFire(ASYNC)，模式为 1，跳过 claim 直接跑。

真正让深链不爆栈的是 postFire 在 mode < 0 时的返回值：

// JDK 25 源码摘录，已删减
final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
    // 前半段先清理源 a 的栈，这里略去
    if (result != null && stack != null) {
        if (mode < 0)
            return this;
        else
            postComplete();
    }
    return null;
}

mode < 0 时它不调用 postComplete，而是把 this 返回给调用者，由外层 while 接着倒；mode >= 0 时才递归调用。于是 postComplete → tryFire(NESTED) → postFire(NESTED) → 返回 是一个平栈循环，只有最初那次 SYNC 触发会多一层。

本机实测（-Xss512k，故意把栈调小）：

场景	结果
手写递归 `1 + down(n-1)`，5000 层	`StackOverflowError`
`thenApply` 链 200000 层	结果 `200000`，耗时 31ms
`thenCompose` 链 200000 层	结果 `200000`，耗时 16ms

耗时是单次运行、随调度波动，不作基准；有意义的是第一行与第二行的对比——同一个 512 KB 栈，手写递归五千层就崩，CF 链二十万层没事。这不是 JIT 的功劳，是 NESTED 把递归改写成了循环。

thenCompose 为什么不走线程

thenCompose 的续体也是栈上的节点。外层完成时 UniCompose.tryFire 调用用户函数拿到内层 future：内层已完成就直接 d.completeRelay(r)；内层未完成就 g.unipush(new UniRelay<V,V>(d, g))，把一个 UniRelay 压到内层 future 的栈上，然后返回 null。

于是"等内层"被表示成"内层的栈上多了一个节点"，而不是"有个线程在 park"。内层完成时 postComplete 弹出 UniRelay，把结果搬到外层，继续往下倒。整条 compose 链的推进力始终是"某个 future 被完成"，没有任何额外调度。

用反射读出真实的栈

stack 和 dep 都是私有字段，下面这段用 --add-opens 读它们——只是实验手段，不进生产代码：

import java.lang.reflect.Field;
import java.util.concurrent.CompletableFuture;

/** 反射读取私有 stack 字段。仅用于实验，生产代码不要这么做。 */
public final class StackProbe {
    static final Field STACK, DEP, NEXT;
    static {
        try {
            STACK = field(CompletableFuture.class, "stack");
            DEP = field(Class.forName("java.util.concurrent.CompletableFuture$UniCompletion"), "dep");
            NEXT = field(Class.forName("java.util.concurrent.CompletableFuture$Completion"), "next");
        } catch (ReflectiveOperationException e) { throw new ExceptionInInitializerError(e); }
    }

    static Field field(Class<?> c, String n) throws NoSuchFieldException {
        Field f = c.getDeclaredField(n); f.setAccessible(true); return f;
    }

    static void dump(String label, CompletableFuture<?>... fs) throws Exception {
        System.out.println("--- " + label + " ---");
        for (CompletableFuture<?> f : fs) {
            Object h = STACK.get(f);
            String out = "#" + Integer.toHexString(System.identityHashCode(f) & 0xffff) + " stack=";
            if (h == null) out += "null";
            for (int i = 0; h != null && i < 8; i++) {
                out += h.getClass().getSimpleName();
                try {
                    if (DEP.get(h) instanceof CompletableFuture<?> d)
                        out += "(dep=#" + Integer.toHexString(System.identityHashCode(d) & 0xffff) + ")";
                } catch (IllegalArgumentException notUni) { out += "(no dep)"; }
                h = NEXT.get(h);
                if (h != null) out += " -> ";
            }
            System.out.println(out);
        }
    }

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> f0 = new CompletableFuture<>();
        CompletableFuture<String> f1 = f0.thenApply(s -> s + "A");
        CompletableFuture<String> f2 = f1.thenApply(s -> s + "B");
        dump("三级 thenApply", f0, f1, f2);

        CompletableFuture<String> g0 = new CompletableFuture<>();
        CompletableFuture<String> inner = new CompletableFuture<>();
        CompletableFuture<String> g1 = g0.thenCompose(s -> inner);
        dump("thenCompose 初始", g0, g1, inner);
        g0.complete("outer");
        dump("外层完成、内层未完成", g0, g1, inner);

        CompletableFuture<String> a = new CompletableFuture<>();
        CompletableFuture<String> b = new CompletableFuture<>();
        CompletableFuture<String> ab = a.thenCombine(b, (x, y) -> x + y);
        dump("thenCombine", a, b, ab);

        CompletableFuture<String> m1 = new CompletableFuture<>();
        CompletableFuture<String> m2 = new CompletableFuture<>();
        CompletableFuture<String> m3 = new CompletableFuture<>();
        CompletableFuture<Void> all = CompletableFuture.allOf(m1, m2, m3);
        dump("allOf 三个源", m1, m2, m3, all);
    }
}

javac --release 25 -d out StackProbe.java
java --add-opens java.base/java.util.concurrent=ALL-UNNAMED -cp out StackProbe

本机输出：

--- 三级 thenApply ---
#c3a8 stack=UniApply(dep=#b418)
#b418 stack=UniApply(dep=#733d)
#733d stack=null
--- thenCompose 初始 ---
#197d stack=UniCompose(dep=#81b5)
#81b5 stack=null
#6ca6 stack=null
--- 外层完成、内层未完成 ---
#197d stack=null
#81b5 stack=null
#6ca6 stack=UniRelay(dep=#81b5)
--- thenCombine ---
#e575 stack=BiApply(dep=#2327)
#acd0 stack=CoCompletion(no dep)
#2327 stack=null
--- allOf 三个源 ---
#905d stack=BiRelay(dep=#16f4)
#b846 stack=CoCompletion(no dep)
#aee7 stack=CoCompletion(no dep)
#ee92 stack=null

读这份输出，有四点源码看不出来的事实：

三级 thenApply 不是一条链表，而是三个各自只有一个节点的栈。 f0 的栈里是 UniApply(dep=f1)，f1 的栈里是 UniApply(dep=f2)，最后一层 f2 的栈是空的。依赖是分散存储的，每个 future 只管自己的直接下游。
thenCompose 之后续体出现在内层 future 的栈上。 外层完成、内层未完成时，inner 的栈顶变成 UniRelay(dep=g1)——“等内层"这个状态被编码成一个节点，而不是一个阻塞的线程。
thenCombine 只创建一个 BiApply，却占用两个栈。 a 的栈上是 BiApply 本身，b 的栈上是一个 CoCompletion 包装。两个源谁先完成都能触发，claim() 的 tag CAS 保证 BiApply 只跑一次。
allOf 建的是归并树，不是一层平铺。 三个源里 m1 上挂 BiRelay（指向中间的归并节点），m2、m3 上各挂一个 CoCompletion。andTree 递归二分，节点数 O(n)、深度 O(log n)。

在 libraryd-conc 里落地

借阅成功后要扇出三个下游：mail、sms、push。用 allOf 聚合，天然就是归并树：

package com.plume.conc.core;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

/** 借阅成功后的通知扇出。三个下游互不阻塞，任一失败不影响其余。 */
public final class NotifyOrchestrator {

    public record Delivery(String channel, boolean ok, String detail) {}

    public interface Notifier {
        Delivery send(Loan loan);
    }

    private final Executor io;
    private final List<String> channels = List.of("mail", "sms", "push");
    private final Map<String, Notifier> notifiers;

    public NotifyOrchestrator(Executor io, Map<String, Notifier> notifiers) {
        this.io = io;
        this.notifiers = notifiers;
    }

    public CompletableFuture<List<Delivery>> fanOut(Loan loan) {
        List<CompletableFuture<Delivery>> parts = channels.stream()
                .map(ch -> CompletableFuture
                        .supplyAsync(() -> notifiers.get(ch).send(loan), io)
                        .exceptionally(ex -> new Delivery(ch, false, ex.getClass().getSimpleName())))
                .toList();
        return CompletableFuture.allOf(parts.toArray(CompletableFuture[]::new))
                .thenApply(ignored -> parts.stream().map(CompletableFuture::join).toList());
    }
}

结构上和上面反射看到的一一对应：三个 supplyAsync 各是一个带 AsyncSupply 的独立 future，每个分支再挂一个 UniExceptionally；allOf 把三个源折成两层 BiRelay。故障隔离不是靠 try/catch 包住整段，而是靠每个分支自己挂一个 UniExceptionally——一个下游异常只让它自己的栈进入异常态，另外两个栈照常触发。

thenApply(ignored -> parts.stream().map(CompletableFuture::join).toList()) 里的 join 不会阻塞：allOf 完成时三个 parts 必然都已完成，这里只是读字段。在已知完成的 future 上 join 是纯读取，在未知完成的 future 上 join 才是阻塞。

这个设计的代价

第一，每个阶段至少两个对象。 一个新 future 加一个 Completion 子类实例；thenCompose 的内层未完成时还要再加一个 UniRelay。那条 20 万层的链分配了大约四十万个对象。CF 适合表达"几十个阶段"的流程，不适合放进每秒百万次的循环——那种场景应该用状态机或显式任务队列。

第二，触发顺序是 LIFO，且不保证跨栈顺序。 同一 future 上的回调逆序执行；两个不同 future 上的回调谁先跑，取决于谁先完成、postComplete 先倒到哪一个。把"先发短信再发邮件"写成两个 thenRun 是错的，必须串联或显式组合。

第三，同步阶段通常跑在完成者的线程上。 thenApply 没有 executor，tryFire 一般就在调用 complete 的线程里跑；13.2 会给出反例——先完成的异步兄弟也会帮忙倒栈。链上任何一个慢函数，都会拖住它所在的那条线程以及接下来要触发的依赖。

第四，Completion 的字段不是 final 也不是 volatile（只有 next 是 volatile）。 源码注释写明它们 “only visible to other threads upon safe publication”。用反射、调试器或 JFR 去读，可能看到只填了一半的节点，所以这类字段不该出现在任何对外契约里。

最后，什么时候不要用这套栈。 需要严格顺序的流程（串成单链后又会回到分配问题）、需要跨阶段传递 ThreadLocal 上下文的（卷一 13.3 讲过 ThreadLocal 不跨线程）、以及纯粹的 CPU 密集短计算（直接算比建两个对象快得多）。

小结

CompletableFuture 用两个 volatile 字段表达全部状态：result 是"完成与否 + 值”，stack 是"等着我的动作"的 Treiber 栈。压栈与完成之间靠"先写自己的标志再读对方的标志"完成无锁握手，unipush 的二次检查保证不丢唤醒。postComplete 把栈倒出来逐个 tryFire(NESTED)，分叉推回根栈、单路径顺着 dep 走，postFire 在 NESTED 下返回 this 而不是递归，于是二十万层的链在 512 KB 栈上也能跑完。thenCompose 的"等待"只是一个 UniRelay 节点，推进力始终来自某个 future 被完成，没有额外的调度线程。

阅读导航：上一节 12.3 并行流的拆分策略｜下一节 13.2 异步任务的执行器选择

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页