`Phaser` 阶段同步器源码

概述

Phaser（阶段同步器，JDK 7 引入）是 CountDownLatch 与 CyclicBarrier 的进阶替代：参与者数量可以动态注册/注销、阶段可以多轮复用、支持树形拓扑。其核心是一个压缩到一个 long 的状态字——phase（阶段）、parties（参与方）、unarrived（未到达数）三位一体，配合 CAS 无锁推进阶段。

本文基于 OpenJDK 21 源码，从 state 位布局出发，拆解注册、到达、等待、阶段推进与树形传播的实现。

核心源码解析

① `Phaser` 的内部状态

java

public class Phaser {
    private volatile long state;     // 位打包状态字
    // 位布局常量
    private static final int  MAX_PARTIES     = 0xffff;      // 参与方上限 65535
    private static final int  MAX_PHASE       = Integer.MAX_VALUE;
    private static final int  PARTIES_SHIFT   = 16;
    private static final int  PHASE_SHIFT     = 32;
    private static final int  UNARRIVED_MASK  = 0xffff;      // 低 16 位
    private static final long PARTIES_MASK    = 0xffff0000L; // 16-31 位
    private static final long TERMINATION_BIT = 1L << 63;    // 最高位：终止标志
    ...
}

位段	位范围	含义
`unarrived`	0-15 位	本轮尚未到达的参与方数
`parties`	16-31 位	已注册参与方总数
`phase`	32-62 位	当前阶段编号（从 0 递增）
`TERMINATION_BIT`	63 位	终止标志（`onAdvance` 返回 true 时置位）

一次 CAS 完成多状态更新：phase/parties/unarrived 打包进一个 long，arrive/register 只需一次 casState 即可原子修改多个维度，避免多字段锁。
MAX_PARTIES = 0xffff 限制了单 Phaser 最多 65535 个参与方；phase 达到 MAX_PHASE 后重置为 0 循环（(phase + 1) & MAX_PHASE）。

② `Phaser.register()` 的分区注册

java

public int register() {
    return doRegister(1);
}
private int doRegister(int registrations) {
    int adj = registrations << PARTIES_SHIFT | registrations;   // 同时增加 parties 与 unarrived
    final Phaser parent = this.parent;
    int phase;
    for (;;) {
        long s = (parent == null) ? state : reconcileState();    // 子节点先与父节点对齐
        int counts = (int)s;
        int parties = counts >>> PARTIES_SHIFT;
        if (registrations > MAX_PARTIES - parties)
            throw new IllegalStateException("Attempted registration count exceeded");
        else if (phase = (int)(s >>> PHASE_SHIFT) < 0)           // 已终止
            break;
        else if (counts == EMPTY) {                              // 首次注册
            if (casState(s, (adj + (int)(s >>> PARTIES_SHIFT)) & COUNTS_MASK))
                break;
        } else if (parent == null || reconcileState() == s) {    // 正常注册
            if (casState(s, s + adj)) break;
        } else if (parent != null) { ... }                       // 父状态变化则重试
    }
    return phase;
}

adj 一次同时抬高 parties 与 unarrived：注册者既算"总参与方"也算"本轮待到达"。
parent 树形注册：带父节点的 Phaser 注册时先 reconcileState() 与父节点对齐阶段，再把注册数传播到父链（root 统一管理）。
返回注册时刻的阶段号，调用方可用它判断是否错过某阶段。

③ `Phaser.arriveAndDeregister()` 的到达退组

java

public int arriveAndDeregister() {
    return doArrive(ONE_DEREGISTER);   // ONE_DEREGISTER = ONE_ARRIVAL | ONE_PARTY
}

private int doArrive(int adjust) {
    final Phaser root = this.root;
    for (;;) {
        long s = (root == this) ? state : reconcileState();
        int phase = (int)(s >>> PHASE_SHIFT);
        if (phase < 0) return phase;                    // 已终止
        int counts = (int)s;
        int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
        if (unarrived <= 0) throw new IllegalStateException(...);
        if (casState(s, s -= adjust)) {                 // ① 原子扣减
            if (unarrived == 1) {                       // ② 最后到达者：推进阶段
                long n = s & PARTIES_MASK;
                int nextUnarrived = (int)n >>> PARTIES_SHIFT;   // 下一轮 unarrived = parties
                if (root != this) return parent.doArrive(nextUnarrived);  // ③ 树形传播
                if (onAdvance(phase, nextUnarrived)) n |= TERMINATION_BIT;
                else if (nextUnarrived == 0) n |= EMPTY;
                else n |= nextUnarrived;
                int nextPhase = (phase + 1) & MAX_PHASE;
                n |= (long)nextPhase << PHASE_SHIFT;
                casState(s, n);                         // ④ 写回新阶段
                releaseWaiters(phase);                  // ⑤ 唤醒等待者
            }
            return phase;
        }
    }
}

ONE_DEREGISTER 同时扣减 unarrived（到达）与 parties（退组）——语义是"我到达本轮且不再参与后续阶段"。
最后到达者负责推进：unarrived 归零时由最后到达的线程执行阶段切换，避免锁唤醒开销。
树形节点把"推进"职责委托给父节点（parent.doArrive(nextUnarrived)），由根节点统一调度。

④ `Phaser.arriveAndAwaitAdvance()` 的到达等待

java

public int arriveAndAwaitAdvance() {
    ...
    if (casState(s, s -= ONE_ARRIVAL)) {
        if (unarrived > 1)
            return root.internalAwaitAdvance(phase, null);   // 还有未到达者：进入等待
        if (root != this) return parent.arriveAndAwaitAdvance();
        ... // 自己是最后到达者：推进并返回新阶段
    }
}

private int internalAwaitAdvance(int phase, QNode node) {
    releaseWaiters(phase - 1);                 // 清理上一阶段的残余等待者
    boolean queued = false;
    int spins = SPINS_PER_ARRIVAL;
    while ((int)((s = state) >>> PHASE_SHIFT) == phase) {   // 阶段未变 → 等待
        if (spins > 0) {                       // ① 先自旋
            --spins;
            Thread.onSpinWait();
        } else if (node == null) {
            node = new QNode(this, phase, false, false, 0L);
        } else if (!queued) {
            queued = tryEnqueue(node);         // ② 加入等待队列
        } else if (node.waiting) {
            LockSupport.park(this);            // ③ 阻塞挂起
        }
    }
    ...
    return phase;                              // 阶段推进后唤醒返回新阶段
}

三级等待策略：先自旋（SPINS_PER_ARRIVAL 次）避免线程切换；自旋耗尽后入 QNode 等待队列；入队后 LockSupport.park 真正阻塞。
QNode 是专用等待节点：携带 phase（等待的哪个阶段）、thread、next 链表指针，供唤醒时精确广播。
releaseWaiters(phase) 在阶段推进后唤醒该阶段的等待者，awaitAdvance(phase) 返回的是新阶段号，调用方通过比较返回值判断是否错过。

⑤ `Phaser.onAdvance(int phase, int registeredParties)` 的阶段转换

java

protected boolean onAdvance(int phase, int registeredParties) {
    return registeredParties == 0;             // 默认：无参与方则终止
}

onAdvance 由最后到达者在推进阶段前调用，返回 true 则置 TERMINATION_BIT，Phaser 进入终止态——此后所有 arrive/register/await 立即返回负数 phase（终止信号）。
默认实现"注册数为 0 即终止"：当所有参与者 arriveAndDeregister 退出后 Phaser 自动关闭，适合"最后一轮任务完成即结束"的场景。
自定义子类可重写 onAdvance 实现业务逻辑：如按阶段数终止（phase == N）、按某条件终止、或打印阶段日志——返回值决定是否终止。
阶段推进写回 state 时 (phase + 1) & MAX_PHASE，phase 溢出后回绕为 0。

⑥ `Phaser` 的树形拓扑

java

public Phaser(Phaser parent) {
    ...
    this.parent = parent;
    this.root  = (parent == null) ? this : parent.root;   // 共享根
    // 根节点维护子节点数组，用于广播阶段推进
    AtomicReference<Phaser>[] subPhasers = this.root.subPhasers;  // root 专用
    ...
}

root 共享：所有子 Phaser 共享同一个根；根节点持有 subPhasers（children 数组），子节点不直接互相感知。
阶段传播：子节点到达时 reconcileState() 读取根的最新 phase；子节点的最后到达者把推进委托给父链，根推进后阶段向整棵树一致。
树形意义：减少单点争用——Phaser 的状态 CAS 集中在根，大量并发子分组各自注册/到达，只在与根对齐阶段时交互；适合"多分组并行 + 统一同步"的大型分治任务（如 ForkJoin 风格框架）。
实现细节：reconcileState() 用 stateForRoot 把子节点的 phase 与根对齐，父状态变化导致 CAS 失败时循环重试。

⑦ `Phaser` vs `CyclicBarrier` / `CountDownLatch`

维度	`CountDownLatch`	`CyclicBarrier`	`Phaser`
计数方向	递减（await 等归零）	固定 N 方到齐	动态 parties + unarrived
复用	一次性	`reset()` 可复用	阶段天然多轮复用
动态注册	不支持	不支持	`register()` / `arriveAndDeregister()`
树形	不支持	不支持	支持（共享 root）
中断/超时	`await(timeout)`	`await(timeout)`	`awaitAdvanceInterruptibly` / timed 变体
完成回调	无	`barrierAction`	`onAdvance`（返回 true 终止）
典型场景	一次性任务栅栏	固定批次的并行汇合	动态参与者、多轮协作、分治同步

选型建议：参与者数量固定且只等一次 → CountDownLatch；固定批次循环协作 → CyclicBarrier；参与者会动态加入/退出、或需要多轮同步 → Phaser。
Phaser 的灵活来自 state 单字段设计：arrive 只关心 unarrived，register 只关心 parties，两者通过位运算互不干扰，同一 CAS 循环内可同时完成"到达 + 退组"。

总结

机制	实现	关键点
状态	`volatile long state`	phase/parties/unarrived 位打包，一次 CAS 全量更新
注册	`doRegister`	`adj` 同时抬高 parties 与 unarrived，树形对齐父阶段
到达	`doArrive`	最后到达者推进阶段、写回新状态、唤醒等待者
等待	`internalAwaitAdvance`	自旋 → QNode 入队 → `LockSupport.park`
终止	`onAdvance`	返回 true 置终止位，phase 变负数
拓扑	`root` + `subPhasers`	子节点委托父链推进，reconcileState 对齐阶段

Phaser 是 JUC 中"单 long 状态机"设计的典范：把三组计数压进一个 volatile long，用 CAS 循环完成所有状态转换，配合自旋降级到阻塞的等待队列与树形传播，兼顾了灵活性、正确性与并发性能。

Phaser 阶段同步器源码 ​

概述 ​

核心源码解析 ​

① Phaser 的内部状态 ​

② Phaser.register() 的分区注册 ​

③ Phaser.arriveAndDeregister() 的到达退组 ​

④ Phaser.arriveAndAwaitAdvance() 的到达等待 ​

⑤ Phaser.onAdvance(int phase, int registeredParties) 的阶段转换 ​

⑥ Phaser 的树形拓扑 ​

⑦ Phaser vs CyclicBarrier / CountDownLatch ​

总结 ​

`Phaser` 阶段同步器源码

概述

核心源码解析

① `Phaser` 的内部状态

② `Phaser.register()` 的分区注册

③ `Phaser.arriveAndDeregister()` 的到达退组

④ `Phaser.arriveAndAwaitAdvance()` 的到达等待

⑤ `Phaser.onAdvance(int phase, int registeredParties)` 的阶段转换

⑥ `Phaser` 的树形拓扑

⑦ `Phaser` vs `CyclicBarrier` / `CountDownLatch`

总结