Java 并发工具类 Phaser 怎么实现动态增减参与线程-java教程-PHP中文网

phaser 是 java 并发包中支持运行时动态注册和注销参与者的同步屏障工具，通过 register() 和 arriveandderegister() 灵活调整未到达计数，适用于分阶段、参与者数量不固定的并行任务。

java 并发工具类 phaser 怎么实现动态增减参与线程

Phaser 是 Java 并发包（java.util.concurrent）中一个灵活的同步屏障工具，相比 CyclicBarrier 和 CountDownLatch，它支持**运行时动态注册和注销参与者**，非常适合分阶段、参与者数量不固定的并行任务场景。

核心机制：通过 register() 和 arriveAndDeregister() 动态调整

Phaser 的“参与线程数”不是构造时固定死的，而是由内部维护的 未到达（unarrived）参与者计数 决定。每次调用：

register()：在当前 phase 中为当前线程增加一个新参与者，内部将 unarrived 计数 +1；
arriveAndDeregister()：当前线程完成本阶段后主动退出，unarrived 计数 −1，且该线程后续不再参与后续 phase 的等待。

注意：调用 register() 不会阻塞，但必须在某个 phase **开始前或进行中**调用才有效；若在 phase 已结束（即已触发栅栏动作）后调用，会直接进入下一个 phase 并注册成功。

Alibabacloud Sdk Client Initialization For Java

在 Java 中初始化和管理阿里云 SDK客户端。包括单例模式、线程安全、endpoint 与 region 配置、VPC 终端节点、同步与异步等。

下载

增减线程的典型模式

常见使用方式包括：

启动时按需注册：主线程创建 Phaser 后不设初始参与者（new Phaser()），各工作线程启动后第一件事就是调用 phaser.register()，确保自己被计入；
分支任务动态加入：某线程在执行中派生子任务（如 ForkJoin 或 new Thread），子任务启动后立即 register()，父线程可继续推进；
提前完成主动退出：某线程判断自己无需参与后续阶段（如数据已处理完毕、条件不满足），调用 arriveAndDeregister() 离开，减少下一阶段等待人数；
批量注销：可通过 phaser.bulkRegister(n) 一次性注册多个参与者（常用于预分配），也可配合 getRegisteredParties() 和 getUnarrivedParties() 监控状态做决策。

关键细节与注意事项

每个线程只能调用一次 register() 进入同一 Phaser 实例（重复调用无效，也不报错）；
arriveAndDeregister() 必须在已注册的前提下调用，否则行为未定义（通常抛 IllegalStateException）；
当 unarrived == 0 时，当前 phase 结束，自动进入下一 phase（phase 数自增），并触发 onAdvance() 回调（可重写）；
如果所有已注册线程都调用了 arriveAndDeregister()，且无新注册，Phaser 会进入终止状态（isTerminated() == true），此后所有操作（除 forceTermination()）均无效；
建议在 finally 块中 deregister，避免因异常导致“幽灵参与者”卡住 phase。

简单示例：动态增减的三阶段处理

假设 2 个初始线程启动，第二阶段有 1 个新线程加入，第三阶段有 1 个线程提前退出：

Phaser phaser = new Phaser();
// 线程 A、B 启动并注册
new Thread(() -> {
    phaser.register(); // 加入
    doWork("A", 1); phaser.arriveAndAwaitAdvance();
    doWork("A", 2); phaser.arriveAndAwaitAdvance();
    doWork("A", 3); phaser.arriveAndDeregister(); // 主动退出
}).start();

new Thread(() -> {
    phaser.register();
    doWork("B", 1); phaser.arriveAndAwaitAdvance();
    doWork("B", 2); phaser.arriveAndAwaitAdvance();
    doWork("B", 3); phaser.arriveAndDeregister();
}).start();

// 第二阶段中途启动新线程 C
new Thread(() -> {
    phaser.register(); // 此时 phase=1，注册后 phase=2 才生效
    doWork("C", 2); phaser.arriveAndAwaitAdvance();
    doWork("C", 3); phaser.arriveAndDeregister();
}).start();

这样，phase 1：2 个参与者；phase 2：3 个；phase 3：仍为 3 个（但 A/B 都 deregister 了，C 还在 —— 实际取决于执行时序，可用 awaitAdvance(int phase) 精确控制）。