Phaser适合多阶段线程协同，支持动态注册与注销，通过arriveAndAwaitAdvance实现阶段同步，结合onAdvance定制各阶段逻辑，适用于参与线程数变化的分步并发场景。
在Java并发编程中，当需要对多个线程按阶段协同执行时，Phaser 是比 CountDownLatch 或 CyclicBarrier 更灵活的选择。它支持动态注册线程、分阶段同步、以及更复杂的协调逻辑。特别适合用于需要分步执行、每阶段参与线程数可能变化的场景。
Phaser的基本概念与核心机制
Phaser 可以看作是 CountDownLatch 和 CyclicBarrier 的结合体，但它支持可变参与线程数和重复使用。每个线程通过 register() 或 arriveAndAwaitAdvance() 加入阶段同步。
关键方法说明：
- arriveAndAwaitAdvance()：通知当前阶段到达，并阻塞等待所有参与者完成
- arriveAndDeregister()：到达但不再参与后续阶段（退出Phaser）
- register()：动态注册一个新参与者
- bulkRegister(n)：批量注册n个参与者
- getPhase()：获取当前阶段编号（从0开始）
模拟多阶段任务处理流程
假设有一个数据处理系统，包含三个阶段：数据加载、数据校验、数据导出。不同阶段参与的线程数量可能不同。
立即学习“Java免费学习笔记（深入）”；
Alibabacloud Sdk Client Initialization For Java
下载
在 Java 中初始化和管理阿里云 SDK客户端。包括单例模式、线程安全、endpoint 与 region 配置、VPC 终端节点、同步与异步等。
public class MultiStageProcessor {
private static final Phaser phaser = new Phaser(1); // 主线程作为控制器
<pre class='brush:java;toolbar:false;'>public static void main(String[] args) {
// 阶段0：启动
System.out.println("系统启动，准备进入第一阶段");
// 第一阶段：数据加载
for (int i = 0; i < 3; i++) {
new Thread(new DataLoader(i)).start();
}
phaser.arriveAndAwaitAdvance(); // 等待所有加载器完成
System.out.println("✅ 第一阶段完成：" + phaser.getPhase());
// 第二阶段：数据校验（新增校验线程）
phaser.register(); // 新增一个校验线程
new Thread(new DataValidator()).start();
phaser.arriveAndAwaitAdvance();
System.out.println("✅ 第二阶段完成：" + phaser.getPhase());
// 第三阶段：数据导出（部分线程退出）
new Thread(new DataExporter()).start();
phaser.arriveAndAwaitAdvance();
System.out.println("✅ 第三阶段完成：" + phaser.getPhase());
phaser.arriveAndDeregister(); // 主线程退出
}
static class DataLoader implements Runnable {
private final int id;
DataLoader(int id) { this.id = id; }
@Override
public void run() {
phaser.register(); // 注册自己
try {
System.out.println("加载器 " + id + " 正在加载数据...");
Thread.sleep(500 + id * 100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
phaser.arriveAndAwaitAdvance();
}
}
}
static class DataValidator implements Runnable {
@Override
public void run() {
try {
System.out.println("校验器正在校验数据...");
Thread.sleep(600);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
phaser.arriveAndAwaitAdvance();
}
}
}
static class DataExporter implements Runnable {
@Override
public void run() {
phaser.register();
try {
System.out.println("导出器正在导出数据...");
Thread.sleep(800);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
phaser.arriveAndAwaitAdvance();
}
}
}}
Phaser的进阶用法与注意事项
Phaser 支持自定义阶段行为，可通过重写 onAdvance(int phase, int registeredParties) 方法实现每阶段结束时的判断逻辑。
Phaser phaser = new Phaser() {
@Override
protected boolean onAdvance(int phase, int registeredParties) {
switch (phase) {
case 0:
System.out.println("第一阶段结束，进入校验");
return false; // 继续下一阶段
case 1:
System.out.println("校验完成，准备导出");
return false;
case 2:
System.out.println("所有任务完成！");
return true; // 返回true则终止Phaser
default:
return true;
}
}
};
使用时注意：
- 每次 arriveAndAwaitAdvance() 调用都会触发 onAdvance 判断是否继续
- 线程应正确调用 register() 和 deregister() 避免计数错误
- 异常处理需谨慎，未捕获异常可能导致线程提前退出而未通知Phaser
- Phaser本身是线程安全的，但共享数据仍需额外同步保护
基本上就这些。Phaser在复杂任务编排中非常实用，尤其适合流水线式或阶段性演进的并发场景。不复杂但容易忽略的是参与者的注册与注销时机，控制好这个就能避免死锁或提前释放问题。
