Java Phaser 如何分阶段协调任务：注册、arrive 与提前终止边界

批处理服务遇到“准备、处理、提交”三段任务时，固定线程数并不难，难的是参与者会在阶段之间变化：某个分片完成后不再参加下一阶段，另一个分片却可能在准备阶段才加入。Java 的 Phaser 把“登记谁要参加”和“谁已经到达”分开处理，适合把这类流程写成可验收的阶段推进。

记住一条主线：先用 register 建立本阶段的参与资格，再用 arrive 或 arriveAndDeregister 报到；最后一个参与者到达时才推进 phase，异常收尾则显式调用 forceTermination。

先把三段批处理映射成三个 phase

下面的示例把三个分片作为初始参与者。每个分片先完成准备，再一起进入处理阶段；分片 2 在处理结束后注销，因此提交阶段只有两个参与者。这个设计的关键不是线程数量，而是每个阶段的 registered parties 是否准确。

import java.util.concurrent.Phaser;

public class PhasedImport {
    static final class ImportPhaser extends Phaser {
        ImportPhaser(int parties) { super(parties); }

        @Override
        protected boolean onAdvance(int phase, int registeredParties) {
            System.out.printf("phase=%d complete, parties=%d%n", phase, registeredParties);
            return registeredParties == 0;
        }
    }

    public static void main(String[] args) throws Exception {
        ImportPhaser phaser = new ImportPhaser(3);
        Thread[] workers = new Thread[3];
        for (int i = 0; i  runShard(phaser, shard));
        }
        for (Thread worker : workers) worker.join();
    }

    static void runShard(Phaser phaser, int shard) {
        phaser.arriveAndAwaitAdvance(); // 准备完成，进入处理
        if (shard == 1) {
            phaser.arriveAndDeregister(); // 分片 1 不参加提交阶段
            return;
        }
        phaser.arriveAndAwaitAdvance(); // 处理完成，进入提交
        phaser.arriveAndDeregister();
    }
}

这里的 new ImportPhaser(3) 已经登记了三个未到达参与者，所以工作线程不需要再次 register。线程启动后第一次调用 arriveAndAwaitAdvance，返回的是它到达时的 phase；最后一个线程到达时，onAdvance 被触发，所有等待者才继续。

动态参与时，register 必须发生在等待之前

如果分片不是启动时就知道，可以在它真正开始工作前调用 register。不要让线程先做一半工作、最后才登记：那样 Phaser 不知道它属于哪个 phase，主流程可能已经提前推进。

int phase = phaser.register();
try {
    prepareShard();
    phaser.arriveAndAwaitAdvance();
    processShard();
    phaser.arriveAndAwaitAdvance();
} finally {
    phaser.arriveAndDeregister();
}

这段写法适合“加入后至少参加两个阶段”的参与者。若任务在准备阶段就失败，不应假装完成处理阶段；应在失败分支中注销自己，并由协调线程决定是否终止整个批次。每个参与者只能为当前 phase 报到一次，否则会出现 IllegalStateException 或阶段计数不符合预期。

onAdvance 与 forceTermination 分别解决什么问题

onAdvance 处理的是“正常阶段刚刚完成后，要不要继续”。示例返回 registeredParties == 0，表示所有参与者都注销后终止。回调中的参数是当前 phase 和推进前的参与者数量；不要在回调里再次注册、到达或等待，Oracle 文档明确说明这些操作的行为不应依赖。

forceTermination 则是外部异常收尾：例如协调线程发现校验结果不可恢复，希望所有等待线程尽快退出。终止后 isTerminated() 为真，后续等待调用会返回负值语义；业务代码仍要在 finally 中清理文件、连接或临时状态。

try {
    validateBatch();
} catch (RuntimeException ex) {
    phaser.forceTermination();
    throw ex;
}
if (phaser.isTerminated()) {
    throw new IllegalStateException("batch terminated");
}

运行时核对四个状态

不要只看线程是否结束。测试时打印 getPhase()、getRegisteredParties()、getArrivedParties() 和 getUnarrivedParties()。正常推进时，未到达数会随报到减少；调用注销后，registered parties 也会减少。若 phase 不动，优先检查是否有参与者漏掉 arrive；若提前终止，检查是否有异常分支调用了 forceTermination。

• 阶段卡住：逐个记录参与者进入和离开每个阶段的日志。
• 阶段跳过：确认动态参与者在本阶段第一次等待前已经 register。
• 注销过早：只有确定不参加后续 phase 时才使用 arriveAndDeregister。
• 回调异常：不要在 onAdvance 内执行阻塞 I/O 或等待其他参与者。

相关问题

Phaser 和 CyclicBarrier 怎么选？

参与者固定且只关心一个屏障时，CyclicBarrier 更直观；参与者需要动态注册、分阶段退出或自定义终止规则时，Phaser 更合适。

arrive 和 arriveAndAwaitAdvance 有什么区别？

arrive 只报到并立即返回；arriveAndAwaitAdvance 还会等待当前 phase 推进，适合阶段之间必须汇合的代码。

为什么 phase 变成负数？

负值表示 Phaser 已终止。检查是否触发了 forceTermination，或者 onAdvance 在参与者归零时返回了 true。

把验收点留在阶段边界

Phaser 的价值在于把“谁参加、谁到达、是否继续”变成可观察状态。实现时先画出每个参与者要经过的 phase，再决定在哪里 register、在哪里 arriveAndDeregister；异常路径则统一由协调方调用 forceTermination。最后用四个查询方法和阶段日志复核，通常比盯着线程池数量更容易找到卡住的那一个参与者。