Java 多线程：Phaser 动态屏障与多阶段任务协作
一、Phaser 的基本原理
1. 为什么需要 Phaser？
之前学习了两个线程同步工具：
CountDownLatch：等待多个任务完成，一次性使用。
CyclicBarrier：多个线程相互等待，所有线程到齐后进入下一阶段，可以重复使用。
但 CyclicBarrier 有一个局限：参与者数量固定，不能动态增加或减少。
例如：
CyclicBarrier barrier = new CyclicBarrier(3);假设有三个线程 A、B、C：
第一阶段：A、B、C 全部参与
第二阶段：A、B、C 全部参与
第三阶段：只有 A、B 参与，C 已经退出此时 CyclicBarrier 仍然需要三个参与者到达屏障。
如果 C 不再调用 await()，A、B 就可能一直等待。
Phaser 正是为了解决这类问题。
2. 什么是 Phaser？
Phaser 是 JDK 提供的阶段同步器，支持多个线程在不同阶段进行同步，并允许参与者动态注册和注销。
Phaser 的核心特点：
支持多个阶段循环使用。
支持动态注册参与者。
支持动态注销参与者。
支持线程报告到达后继续执行其他操作。
支持中断、超时和强制终止。
支持通过 onAdvance() 定义阶段完成后的行为。
Phaser 是 Java 7 开始提供的类，Java 8 可以直接使用，不需要额外依赖。
3. Phaser 的三个核心概念
phase：阶段编号
表示当前正在执行哪个阶段。
初始阶段：phase = 0
第一轮完成：phase = 1
第二轮完成：phase = 2正常情况下，每完成一轮同步，阶段编号递增。
registeredParties：已注册参与者数量
例如：
Phaser phaser = new Phaser(3);表示初始注册三个参与者。
unarrivedParties：当前阶段尚未到达的参与者数量
例如：
初始：
registeredParties = 3
unarrivedParties = 3
线程 A 到达：
registeredParties = 3
unarrivedParties = 2
线程 B 到达：
registeredParties = 3
unarrivedParties = 1
线程 C 到达：
当前阶段完成，进入下一阶段正常进入下一阶段后，unarrivedParties 会根据新的参与者数量重新计算。
注意：
Phaser 注册的是逻辑参与者，并不是自动绑定某个 Thread 对象。
同一个线程可以代表多个参与者，但必须自行保证注册、到达和注销次数正确。
二、Phaser 的核心方法
1. register()：注册参与者
Phaser phaser = new Phaser();
phaser.register();
phaser.register();
phaser.register();此时：
registeredParties = 3
unarrivedParties = 3也可以直接初始化：
Phaser phaser = new Phaser(3);或者批量注册：
Phaser phaser = new Phaser();
phaser.bulkRegister(3);在正在推进阶段的并发环境中，注册操作也可能等待当前推进完成，因此不要假设它在任何情况下都绝对不阻塞。
2. arrive()：只报告到达，不等待
int phase = phaser.arrive();作用是报告当前参与者已经完成本阶段。
例如：
初始：
registeredParties = 3
unarrivedParties = 3
线程 A 调用 arrive()
结果：
registeredParties = 3
unarrivedParties = 2特点：
减少当前阶段未到达的参与者数量。
不等待其他参与者。
不会注销当前参与者。
当前参与者仍然需要参与下一阶段。
注意：arrive() 返回的是到达时对应的阶段编号。
3. arriveAndAwaitAdvance()：到达并等待
phaser.arriveAndAwaitAdvance();作用：
当前参与者报告到达。
等待其他参与者到达。
当前阶段全部完成后，进入下一阶段。
例如：
线程 A：arriveAndAwaitAdvance() → 等待
线程 B：arriveAndAwaitAdvance() → 等待
线程 C：arriveAndAwaitAdvance() → 最后到达
所有线程进入下一阶段它的行为类似 CyclicBarrier.await()。
4. arriveAndDeregister()：到达并注销
phaser.arriveAndDeregister();作用：
报告当前阶段到达。
减少已注册参与者数量。
后续阶段不再等待该参与者。
例如：
初始：
registeredParties = 3
unarrivedParties = 3
线程 C 调用 arriveAndDeregister()
结果：
registeredParties = 2
unarrivedParties = 2如果其他两个线程完成当前阶段，下一阶段就只需要等待 A 和 B。
这是 Phaser 相比 CyclicBarrier 最重要的优势。
5. awaitAdvance()：只等待阶段变化
int phase = phaser.arrive();
phaser.awaitAdvance(phase);这里发生了两件事：
arrive()
|
v
报告本阶段到达
|
v
继续执行其他操作
|
v
awaitAdvance(phase)
|
v
等待阶段推进
|
v
进入下一阶段两种写法可以实现相似的同步效果：
第一种：
phaser.arriveAndAwaitAdvance();第二种：
int phase = phaser.arrive();
phaser.awaitAdvance(phase);区别是第二种允许把到达和等待分开执行。
需要特别记住：
awaitAdvance() 不会减少 unarrivedParties，只有 arrive() 等到达方法才会报告到达。
三、Java 8 实战：动态参与者与多阶段任务
1. 业务场景
假设有三个任务：
线程 A：执行三个阶段。
线程 B：执行三个阶段。
线程 C：只执行前两个阶段。
执行要求：
第一阶段：
A、B、C 全部完成
|
v
进入第二阶段
第二阶段：
A、B、C 全部执行
|
v
C 完成后注销
|
v
进入第三阶段
第三阶段：
只有 A、B 执行
|
v
全部完成2. 完整代码
import java.util.concurrent.Phaser;
public class PhaserDemo {
public static void main(String[] args)
throws InterruptedException {
// 主线程先注册，防止任务注册过程中屏障提前推进
Phaser phaser = new Phaser(1) {
@Override
protected boolean onAdvance(
int phase,
int registeredParties) {
System.out.println(
"第 " + (phase + 1)
+ " 阶段完成，剩余参与者："
+ registeredParties
);
// 所有参与者注销后终止
return registeredParties == 0;
}
};
Thread[] workers = new Thread[3];
for (int i = 1; i <= 3; i++) {
final int taskId = i;
// 为当前任务注册参与者
phaser.register();
workers[i - 1] = new Thread(() -> {
boolean registered = true;
try {
// 第一阶段：A、B、C
executeStage(taskId, 1);
if (phaser.arriveAndAwaitAdvance() < 0) {
return;
}
// 第二阶段：A、B、C
executeStage(taskId, 2);
if (taskId == 3) {
// 线程 C 第二阶段完成后退出
phaser.arriveAndDeregister();
registered = false;
System.out.println(
"线程 C 已退出后续阶段"
);
return;
}
if (phaser.arriveAndAwaitAdvance() < 0) {
return;
}
// 第三阶段：只有 A、B
executeStage(taskId, 3);
// 完成后注销
phaser.arriveAndDeregister();
registered = false;
} catch (InterruptedException e) {
// 某个任务中断，终止整个协作
phaser.forceTermination();
Thread.currentThread().interrupt();
} catch (RuntimeException e) {
phaser.forceTermination();
throw e;
} finally {
// 未正常注销且未终止时，清理注册资格
if (registered && !phaser.isTerminated()) {
phaser.arriveAndDeregister();
}
}
}, "worker-" + taskId);
workers[i - 1].start();
}
// 所有任务注册后，主线程退出屏障
phaser.arriveAndDeregister();
// 等待所有线程结束
for (Thread worker : workers) {
worker.join();
}
System.out.println("全部任务处理结束");
}
private static void executeStage(
int taskId,
int stage) throws InterruptedException {
System.out.println(
"线程 " + taskId
+ " 执行第 " + stage + " 阶段"
);
Thread.sleep(taskId * 200L);
}
}3. 执行过程
第一阶段：
phase = 0
参与者：A、B、C
registeredParties = 3三个线程全部到达：
phase = 1第二阶段：
参与者：A、B、C线程 C 调用：
phaser.arriveAndDeregister();注册人数减少：
registeredParties = 2A、B 也完成第二阶段后：
phase = 2第三阶段：
参与者：A、B
registeredParties = 2A、B 完成后注销。
所有参与者数量变为 0，onAdvance() 返回 true，Phaser 终止。
这里主线程先注册为一个临时参与者，是为了确保三个工作线程全部完成注册后，才允许第一阶段开始推进。
4. onAdvance() 的作用
Phaser 提供：
protected boolean onAdvance(
int phase,
int registeredParties
);它会在当前阶段所有参与者到达、准备推进阶段时执行。
参数含义：
phase：刚刚完成的阶段编号。
registeredParties：此时已注册的参与者数量。
返回值：
false：继续进入下一阶段。
true：终止 Phaser。
例如：
Phaser phaser = new Phaser(3) {
@Override
protected boolean onAdvance(
int phase,
int registeredParties) {
// 完成三个阶段后终止
return phase >= 2 || registeredParties == 0;
}
};这里会在编号为 2 的阶段完成时终止，也就是总共完成三个阶段。
注意：
不建议在 onAdvance() 中执行耗时较长的数据库查询、网络请求等操作，因为会延迟所有等待参与者进入下一阶段。
四、Phaser 的中断、超时与终止机制
1. arriveAndAwaitAdvance() 能响应中断吗？
phaser.arriveAndAwaitAdvance();这个方法不会因为中断就直接抛出 InterruptedException 并放弃等待。
它属于不可中断的阶段等待方式。
如果需要可中断等待，应该使用：
int phase = phaser.arrive();
try {
phaser.awaitAdvanceInterruptibly(phase);
} catch (InterruptedException e) {
phaser.forceTermination();
Thread.currentThread().interrupt();
}注意：
arrive() 已经报告了本阶段到达。
因此后面只需要 awaitAdvanceInterruptibly()，不能再调用 arriveAndAwaitAdvance() 重复报告。
2. 如何设置等待超时？
可以使用：
int phase = phaser.arrive();
try {
phaser.awaitAdvanceInterruptibly(
phase,
5,
TimeUnit.SECONDS
);
} catch (InterruptedException e) {
phaser.forceTermination();
Thread.currentThread().interrupt();
} catch (TimeoutException e) {
phaser.forceTermination();
System.out.println("等待阶段推进超时");
}需要引入：
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;它们都是 Java 8 自带的类。
这里有一个非常重要的区别：
Phaser 等待超时或中断，不会自动让整个 Phaser 终止。
而 CyclicBarrier 中，某个等待线程超时或中断，通常会使当前屏障进入 broken 状态。
Phaser 如果需要整个任务组终止，必须通过业务代码自行协调。
3. forceTermination() 是什么？
phaser.forceTermination();作用是强制终止 Phaser。
终止后：
Phaser 不再正常推进阶段。
等待阶段变化的线程可以结束等待。
isTerminated() 返回 true。
getPhase() 返回负数。
新的注册不再正常增加参与者。
注意：
forceTermination() 不是 Thread.stop()，不会直接终止正在执行业务的线程。
例如：
线程 A：等待 Phaser
线程 B：正在执行数据库查询
线程 C：发生异常，调用 forceTermination()此时：
A 可以从阶段等待中返回。
B 不会因为 Phaser 终止，就立即停止数据库查询。
B 需要主动检查终止状态或配合其他取消机制。
因此，生产环境中还需要做好任务取消和异常处理。
五、Phaser 与 CountDownLatch、CyclicBarrier 的区别
1. CountDownLatch
适合：
主线程等待三个子任务结束。例如：
CountDownLatch latch = new CountDownLatch(3);三个任务分别 countDown()，主线程 await()。
2. CyclicBarrier
适合：
三个线程每个阶段都要到齐。例如：
CyclicBarrier barrier = new CyclicBarrier(3);三个线程每一阶段都调用 await()。
3. Phaser
适合：
第一阶段：三个线程
第二阶段：五个线程
第三阶段：两个线程可以使用：
phaser.register();
phaser.arriveAndAwaitAdvance();
phaser.arriveAndDeregister();动态控制参与者数量。
选择原则：
等待若干任务结束：CountDownLatch。
固定线程数量、循环分阶段：CyclicBarrier。
动态线程数量、复杂多阶段协作：Phaser。
六、生产环境常见问题
1. arrive() 后重复到达
错误：
int phase = phaser.arrive();
// 错误：同一参与者再次报告当前阶段到达
phaser.arriveAndAwaitAdvance();这可能造成到达计数错误，甚至抛出 IllegalStateException。
正确：
int phase = phaser.arrive();
phaser.awaitAdvance(phase);或者：
phaser.arriveAndAwaitAdvance();重点：
同一个注册参与者在同一个阶段通常只能报告一次到达。
2. 注册后忘记注销或报告到达
例如：
phaser.register();
executeTask();如果 executeTask() 发生异常，后续没有 arrive() 或注销操作，其他线程可能一直等待。
因此必须明确失败处理策略：
正常完成时报告到达。
不再参与后续阶段时注销。
整个协作无法继续时终止 Phaser。
3. 线程池容量不足
例如：
ExecutorService executor =
Executors.newFixedThreadPool(2);
Phaser phaser = new Phaser(3);提交三个需要在同一屏障等待的任务。
可能发生：
线程池只有两个工作线程
任务 A：进入阶段等待
任务 B：进入阶段等待
任务 C：还在队列中，无法开始如果没有超时或终止机制，就可能一直无法推进。
与 CyclicBarrier 一样，Phaser 也需要合理安排线程池容量与任务调度方式。
4. Phaser 不管理业务执行结果
Phaser 只负责阶段同步。
例如：
executeTask();
phaser.arriveAndAwaitAdvance();如果 executeTask() 失败，Phaser 不会自动记录任务失败原因。
实际业务中，还需要通过 Future、结果容器或其他机制收集异常。
5. 不要错误理解参与者和线程的关系
Phaser 不会自动绑定线程身份。
例如：
phaser.register();并不是将当前线程永久绑定为一个参与者。
它只是增加逻辑参与者数量。
因此，需要业务代码正确维护注册、到达和注销次数。
6. Phaser 不是分布式同步工具
Phaser 只能协调当前 JVM 内共享同一个 Phaser 实例的线程。
多个 Docker 容器之间不能直接共享 Phaser。
如果需要跨实例协调任务阶段，需要使用数据库、Redis、消息队列等分布式机制。
七、重点总结
需要重点记住：
Phaser 可以理解为支持动态参与者的多阶段同步工具。
Phaser 的阶段编号从 0 开始。
register() 增加参与者。
arrive() 只报告到达，不等待。
arriveAndAwaitAdvance() 报告到达并等待。
arriveAndDeregister() 报告到达并注销。
awaitAdvance() 不会减少未到达数量。
同一个注册参与者在同一阶段不能重复报告到达。
Phaser 支持循环使用，但终止后不能继续正常推进。
Phaser 等待超时或中断，不会自动终止整个 Phaser。
forceTermination() 不会直接停止正在执行业务的线程。
Phaser 不直接继承 AQS。
Phaser 只用于 JVM 内部线程同步，不能直接实现分布式阶段协调。
八、常见问题与答案
1. Phaser 和 CyclicBarrier 最大的区别是什么？
Phaser 可以动态注册和注销参与者，CyclicBarrier 的参与者数量固定。
2. Phaser 为什么支持多阶段同步？
因为它维护阶段编号和当前阶段的参与者状态，所有参与者到达后可以推进到下一阶段。
3. arrive() 和 arriveAndAwaitAdvance() 有什么区别？
arrive() 只报告到达，不等待；arriveAndAwaitAdvance() 报告到达后会等待其他参与者。
4. arrive() 与 awaitAdvance() 为什么可以组合使用？
因为前者负责报告到达，后者负责等待阶段变化，两个操作可以分开执行。
5. arriveAndDeregister() 有什么作用？
报告当前阶段到达，同时注销参与者，后续阶段不再等待它。
6. Phaser 的 phase 为什么从 0 开始？
0 代表初始阶段，每完成一轮正常同步，阶段编号会推进。
7. Phaser 可以重复使用吗？
可以。正常阶段完成后可以继续下一轮，直到终止。
8. Phaser 等待超时会导致所有线程一起失败吗？
不会自动导致整个 Phaser 终止，需要业务主动处理失败和取消。
9. onAdvance() 是什么？
它是阶段推进时执行的回调，可以执行简单的阶段结束操作，也可以通过返回值决定是否终止。
10. Phaser 为什么不能代替 Future？
Phaser 主要负责阶段同步，不负责保存异步任务结果和异常；Future 可以用于获取异步任务执行结果。
11. 为什么 Phaser 的注册参与者不等同于线程？
因为 Phaser 只管理逻辑参与者计数，不会自动记录某个线程是否拥有注册资格。
12. Phaser 适合什么场景？
动态增加或减少参与者、需要多阶段同步的大型并行任务，例如分阶段数据处理、批量计算等。
九、复习测试
Phaser 是什么？主要解决什么问题？
Phaser 和 CyclicBarrier 最大的区别是什么？
phase、registeredParties、unarrivedParties 分别代表什么？
register() 和 bulkRegister() 有什么区别？
arrive() 和 arriveAndAwaitAdvance() 有什么区别？
arriveAndDeregister() 为什么适合动态参与者退出？
awaitAdvance() 会不会减少未到达参与者数量？
为什么 arrive() 后不能立即再次 arriveAndAwaitAdvance()？
Phaser 的阶段编号什么时候增加？
onAdvance() 的返回值代表什么？
Phaser 为什么支持重复使用？
arriveAndAwaitAdvance() 是否支持中断取消等待？
如何实现可中断、可超时的阶段等待？
Phaser 超时后为什么可能需要调用 forceTermination()？
forceTermination() 会不会直接停止其他业务线程？
CountDownLatch、CyclicBarrier、Phaser 应该如何选择？