Phaser 这个东西我在做多阶段并行任务编排的时候第一次真正用上之前一直靠 CountDownLatch 和 CyclicBarrier 硬凑凑到最后代码里飘着五六个计数器和栅栏对象改一次阶段划分就要动五六处地方改完还得重新推一遍并发路径。java.util.concurrent.Phaser是 JUC 包里专门做多任务协同的同步器说白了就是一个能动态增减参与方、能连续推进多个阶段、还能在每道关口上挂回调的栅栏。它解决的是很具体的一类问题一批任务要分好几轮跑每一轮结束所有任务必须对齐对齐之后才能进下一轮而且参与方数量在运行期还会变有的分片提前干完要退出有的临时要加人帮忙。这种活儿用 CountDownLatch 做会非常别扭因为它的计数只能减不能加用完一次就得换新的。适合谁看写过线程池、用过 CountDownLatch 或 CyclicBarrier现在想把并行流程做得更细的同学完全没接触过并发工具也能看我会先把概念拆开讲再上代码。1. 先搞清楚 Phaser 顶上的是哪块拼图1.1 CountDownLatch 和 CyclicBarrier 卡在哪CountDownLatch 的模型是一根单向的绳子。构造的时候给它一个数谁干完活谁就 countDown 一次等到归零所有卡在 await 上的线程一起放行。它的优点是最简单、语义最直白缺点是三个计数只能减不能加、用完一次就废了、没有第几轮的概念。你如果要做三轮都对齐的流程就得 new 三个 CountDownLatch而且每一轮的线都要重新串一遍代码里到处都是 latch1、latch2、latch3。更麻烦的是如果某一轮执行到一半发现有分片失败不再参与了你得手动把那部分计数补掉否则所有人一起等死。CyclicBarrier 像是固定人数的集合点一车人到了就发车发完车原地重置下一轮继续。它解决了循环使用的问题但它有两个硬约束。第一个是参与方数量在构造时定死运行期加不进人也减不掉人——你不能说这个任务今天不来了只能让它照样 await或者干脆打破整个屏障。第二个是它内部只有一个 generation 对象没有对外暴露阶段号你想知道现在是第几轮只能自己拿个原子变量在外面数。另外它有个挺讨厌的特性任何一个人 await 时被打断屏障就 Broken其他人会收到 BrokenBarrierException然后你要在业务代码里处理这个异常并决定是重试还是放弃写起来不轻松。我在一个批次作业里就吃过这个亏12 个分片跑到第 3 轮的时候有一个分片因为数据格式问题抛了异常直接退出剩下的 11 个线程全部卡在 await 上JVM 一直不退出最后只能人工重启。那次之后我就开始认真看 Phaser 的文档了。1.2 Phaser 多出来的四件事Phaser 的定位是把 CountDownLatch 和 CyclicBarrier 的能力合并并且把两个硬约束拆掉。具体来说它多出来四件事这四件事正好对应上面那些坑。第一参与方可以在运行期动态注册和注销。register()加一个bulkRegister(int)批量加arriveAndDeregister()到场并退出。这意味着某个分片干完三轮就不干了是一等公民操作不需要额外的簿记逻辑。第二有明确的阶段号phase。phase 从 0 开始每推进一轮加一你可以随时getPhase()读出来也可以用awaitAdvance(int)等一个指定的阶段号。有了这个编号第几轮这件事就不用自己在外面数了。第三推进规则可以自定义。覆写onAdvance(int phase, int registeredParties)就能决定这一轮结束后是继续还是收摊返回值是 true 就终止。这让跑够 4 轮就结束、没人了自动收摊、出错立刻熔断都能写在一个地方。第四支持分层tiering。几个子 Phaser 可以挂到同一个父 Phaser 上形成两级甚至多级屏障。参与方上百之后这一招能明显减少单点竞争。下面这张表是我自己选型时会对照的直接拿去用。对比维度CountDownLatchCyclicBarrierPhaser计数方向只减不加每轮自动重置可增可减能否循环使用不能一次性能能直到终止参与方能否运行期变更不能不能能register / arriveAndDeregister是否有阶段编号无内部有 generation不对外暴露数字有getPhase 可读是否支持分层不支持不支持支持父子 Phaser等待是否响应中断await 响应await 响应但会导致 BrokenarriveAndAwaitAdvance 不响应awaitAdvanceInterruptibly 响应终止方式计数归零自然释放到达即重置由 onAdvance 返回值决定典型场景启动前等外部依赖就绪固定人数多轮同步阶段化流水线、动态参与方、分片并行注意这张表里等待是否响应中断那一行是我吃过亏的地方。arriveAndAwaitAdvance()在中途被 interrupt 之后不会抛异常它会继续等下去。这意味着线上想靠中断来撤销一个卡住的任务是没用的只能靠forceTermination()。2. 机制拆解phase 是怎么一步一步往前推的2.1 一个 long 字段里装下的全部状态Phaser 内部没有一个庞大的对象图它把所有状态压在一个volatile long state里。以 JDK 8 的实现为例这 64 位是这么切的低 16 位是还没到场的参与方数量第 16 到 31 位是已注册的参与方总数第 32 到 62 位是当前 phase 号最高位是终止标记。这个设计带来的直接推论有三条都很实用。第一条参与方总数上限是 65535。超过这个数register()会直接抛IllegalStateException异常信息里会带上当前 state 的十六进制值。这个数字对绝大多数业务来说够用但如果你的任务粒度切得特别细几千上万个微任务各占一个 party就会撞到。第二条phase 是一个 32 位有符号整数理论上跑到Integer.MAX_VALUE之后会回绕。实际业务里跑到几十万轮已经很夸张了但如果真有人拿 Phaser 做一个 7x24 小时不断推进的事件循环那确实需要在外部兜一层到一定轮数就重建 Phaser。我见过一个常驻仿真服务就是每 100 万轮换一个新的 Phaser成本很低规避了所有边角问题。第三条所有状态变更都是 CAS。arrive()的本质就是把这个 long 里的未到场数减一减到零的那个线程负责推进 phase。因为只有一个字所以这些操作相对轻量短阶段的情况下很多线程甚至不需要真正阻塞自旋几次就过去了。JDK 里有个SPINS_PER_ARRIVAL常量控制自旋次数跟 CPU 核数相关这也是 Phaser 在细粒度阶段上比 CyclicBarrier 更轻的原因之一。2.2 最后一个到达的线程负责推进阶段Phaser 的推进模型是最后一个到场的人关门。假设当前 phase 是 2注册了 8 个参与方。前 7 个调用arriveAndAwaitAdvance()时它们只是把未到场数从 8 减到 1然后自己进队列挂起。第 8 个调用的时候未到场数减到 0它就承担了推进者的角色先把 phase 加一再调用onAdvance(2, 8)然后唤醒队列里等着的 7 个线程最后自己返回。这个模型解释了两个现象。一是阶段推进的日志只会打一次因为推进只有一个线程做。二是在onAdvance里做重活儿会拖慢整组因为推进者必须先干完你写的活儿才去叫醒其他人。我之前在onAdvance里写过一个把各分片统计结果汇总写进数据库的逻辑结果每轮多花 200 毫秒8 个分片全部陪着等吞吐直接掉了一截。后来改成在onAdvance里只做一次快照volatile引用替换落库交给单独的线程问题就没了。返回值这块也有个容易忽略的点。arriveAndAwaitAdvance()返回的是推进之后的新 phase 号如果 Phaser 已经终止返回的是负数。所以在循环里可以直接用返回值判断要不要继续int next phaser.arriveAndAwaitAdvance(); if (next 0) { // 已经终止收工 break; }这比在循环条件里反复isTerminated()要可靠因为isTerminated()和你的决策之间存在时间窗口。2.3 onAdvance 是唯一能改写终止规则的地方不覆写onAdvance的话默认实现就一行逻辑return registeredParties 0;。翻译过来就是没人了就收摊。这个默认行为在任务全部用 arriveAndDeregister 退出的模式下很方便但它有两个副作用必须记住。第一个副作用是一旦 Phaser 进入终止状态它就彻底死了。register()不会抛异常而是返回一个负数表示这次注册没有任何效果arrive()和arriveAndDeregister()也直接返回负数。这里最阴的地方在于 register 的静默失败如果你的代码是先注册再提交任务而注册的时候 Phaser 已经终止了那么这个任务实际上不是一个参与方屏障不会等它。程序不报错但结果偶尔少一块这种 bug 排查起来非常费劲。我的做法是永远检查返回值int p phaser.register(); if (p 0) { throw new IllegalStateException(Phaser 已终止任务无法注册为参与方); }第二个副作用是空 Phaser 不能当待机用。new Phaser(0)之后如果发生一次推进默认 onAdvance 会因为注册方为 0 直接判定终止。如果你想先建对象后加人正确做法是把初始 parties 设成 1由主控线程先占一个位最后再用arriveAndDeregister()退出这也是最常用的发令枪模式。覆写的时候按需求挑条件就行下面是我常用的三种写法。// 写法一跑够固定轮数就收摊 Phaser phaser new Phaser(SHARDS) { Override protected boolean onAdvance(int phase, int registeredParties) { return phase 1 TOTAL_STAGES || registeredParties 0; } }; // 写法二不允许中途有人退出谁退谁终止适合强一致场景 Phaser strict new Phaser(SHARDS) { Override protected boolean onAdvance(int phase, int registeredParties) { return registeredParties SHARDS; } }; // 写法三只在全员自然退出时终止阶段数不设上限 Phaser openEnded new Phaser(SHARDS);实操心得phase 1 TOTAL_STAGES里的1特别容易漏。因为onAdvance收到的是即将结束的那一轮的编号从 0 开始。想跑 4 轮条件是phase 1 4写成phase 4就会多跑一轮。2.4 分层 Phaser参与方上百之后要考虑的事单个 Phaser 上参与方一旦上到几百CAS 冲突就会变得明显所有线程都在抢同一个 long自旋次数上升实际表现为阶段推进的耗时随参与方数量非线性增长。我实测过 32 个分片和 512 个分片跑同样的阶段逻辑后者的单阶段开销大概是前者的 20 倍以上远超线性。分层的思路是把一个大屏障拆成两级。创建若干个小组每组挂一个子 Phaser子 Phaser 的父指针指向根 Phaser。组内的线程只在子 Phaser 上抢组与组之间通过根 Phaser 对齐。根上只有组数 主控这么几个参与方竞争压力降下来一个数量级。// 根 Phaser 先给主控占一个位 Phaser root new Phaser(1); // 两组每组 64 个 worker Phaser groupA new Phaser(root, 64); Phaser groupB new Phaser(root, 64);用分层有两个必须知道的行为。第一个是父子阶段是对齐的子组的线程即使在子 Phaser 上到齐了也得等根 Phaser 推进之后才能继续往下走。也就是说根没动子组出不去。第二个是子 Phaser 终止时会自动从父 Phaser 注销所以正常退出不需要手动维护父子关系。我一般只在参与方超过 128 个的时候才考虑分层。低于这个数分层的收益还不够抵消多一层抽象的复杂度调试的时候看栈也更容易看晕。3. 实操把一条四阶段流水线跑起来3.1 场景设定和阶段划分为了讲得具体我拿一个离线分片作业当例子这套结构在实际项目里很常见。作业把一个大批次拆成 8 个分片每个分片要跑 4
优化到O(n²))
