Java进阶篇之Phaser：让参与者灵活加入与退出阶段协作

上一篇的CyclicBarrier适合同一组工作者反复会合。再往前走一步如果准备阶段需要三位工作者后续加工只需要其中两位怎样让队伍在阶段之间自然调整今天认识Phaser把注册、到达和退出放进一个完整例子里。本文以JDK21为准Phaser自Java7提供官方资料核对日期为2026年9月23日。例子使用虚构的批处理流程不涉及真实业务数据。一、先看结构阶段与参与份额把一次批处理拆成准备、加工两个阶段。每个仍在队伍中的工作者本阶段完成后报到当前阶段需要的份额全部报到阶段才会推进。阶段编号从0开始而注册份额可以随着流程变化。JDK21的Phaser说明Phaser只管理计数并不保存“哪一个线程已经注册”的名单。因此代码必须自己保证注册和到达次数一致。把一个份额理解成一项持续参与的责任比简单地数线程更容易检查流程。本文安排A、B、C三个工作者A和B准备完后等待C只负责准备完成后退出。此时第一阶段齐了A和B可以开始加工后续不用再等待C。图中的卡片代表参与份额猫是讲解者不表示线程身份。C退出的同时也完成了本阶段的报到。二、把几个动作分清楚动作方法理解方式增加参与份额register()当前流程又多了一项到达责任到达后等待arriveAndAwaitAdvance()本阶段工作完成等大家齐了再继续到达但不等待arrive()只报到是否等待另行安排到达并退出arriveAndDeregister()本阶段报到后续不再参加按阶段等待awaitAdvance(phase)等指定阶段发生变化这些方法的语义依据JDK21 API。实际使用时建议先画出每个工作者在哪一阶段离开再写代码。尤其不要把arrive()和随后一次arriveAndAwaitAdvance()连着调用来“补一个等待”后者会再次报到计数就乱了。三、完整示例三人准备两人加工这里先一次性注册三个份额再启动三个线程。示例故意不使用随机休眠让协调逻辑保持清晰。importjava.util.concurrent.Phaser;publicclassPhaserDemo{publicstaticvoidmain(String[]args)throwsInterruptedException{PhaserphasernewPhaser(3);Threadaworker(A,false,phaser);Threadbworker(B,false,phaser);Threadcworker(C,true,phaser);a.start();b.start();c.start();a.join();b.join();c.join();System.out.println(全部结束已终止phaser.isTerminated());}privatestaticThreadworker(Stringname,booleanpreparationOnly,Phaserphaser){returnnewThread(()-{System.out.println(name准备完成);if(preparationOnly){phaser.arriveAndDeregister();return;}intnextPhasephaser.arriveAndAwaitAdvance();if(nextPhase0){return;}System.out.println(name加工完成);phaser.arriveAndDeregister();},worker-name);}}保存为PhaserDemo.java可用javac PhaserDemo.java编译再运行java PhaserDemo。按代码逻辑一种可能的输出如下同阶段内部的先后顺序并不固定A准备完成 C准备完成 B准备完成 B加工完成 A加工完成 全部结束已终止true观察输出时抓住两个约束即可两条“加工完成”前三条“准备完成”都已经出现最后一行在三个线程结束后出现。这里主线程靠**join()**等待线程生命周期结束工作者之间靠Phaser协调阶段两个等待点各有职责。四、动态注册时先留住协调份额如果任务列表是运行时才确定的可以让协调线程先占一个份额避免前面的任务过早结束、整个协调器已经终止后面的任务才来注册。PhaserphasernewPhaser(1);// 协调者先占一个份额for(Runnabletask:tasks){phaser.register();// 注册在提交前提交失败时也必须处理这份责任try{executor.execute(()-{try{task.run();}finally{phaser.arriveAndDeregister();}});}catch(java.util.concurrent.RejectedExecutionExceptionex){phaser.arriveAndDeregister();// 示例选择记录并继续生产代码应明确失败策略System.err.println(任务提交失败ex.getMessage());}}phaser.arriveAndAwaitAdvance();phaser.arriveAndDeregister();这段是局部用法tasks和executor由调用方提供。它等待已提交任务结束却没有汇总业务成功与否。需要结果时可额外保存Future或失败记录“到齐”与“全部成功”应分别判断。五、容易忽略的注意事项**每阶段只报到一次。**如果已经用arrive()报到后面需要的是等待操作异常清理时也要结合当前阶段责任判断不能无条件再减一次。**超时不替你移除参与者。**需要可中断或限时等待时可使用awaitAdvanceInterruptibly。等待异常不会自动破坏PhaserforceTermination()能释放等待者但不会自动取消业务任务。默认情况下全部份额退出后会终止之后重新注册无法恢复。官方等待与终止语义**线程池容量要覆盖等待依赖。**如果先运行的任务占满线程池并等待尚在队列里的参与者流程就可能无法推进。动态计数解决不了执行资源不足的问题。六、思维导图Phaser阶段协作注册参与份额每阶段报到一次参与者到齐后推进到达并退出后续阶段等待异常需单独处理业务结果与协调状态分开总结要点Phaser适合参与数量会变化的阶段协作。先明确每个工作者承担几轮责任再把注册、到达和退出对应到流程里代码会更容易核对。到达次数正确是阶段正常推进的基础。对动态任务要同时处理提交失败、执行失败和等待超时避免留下无人履行的份额。协调工具只负责推进时机。业务是否成功、线程是否结束、失败后如何取消仍然需要各自清楚的处理方式。下一篇继续认识Semaphore看看怎样限制同时进入某项资源的任务数量。如果你觉得这篇文章对你有所帮助欢迎点赞、收藏、分享