内容概要
Phaser是Java中一个灵活的同步工具,其优点在于支持多阶段的任务拆分与同步,并且能够动态地注册与注销参与者,它提供了丰富的等待与推进机制,使得开发者能够更细粒度地控制线程的协调行为,实现复杂的并行任务处理,相比于其他同步工具,Phaser更加灵活且易于扩展,适用于多种并发场景。
核心概念
在Java中,Phaser
是一个灵活的同步工具类,它允许多个线程在一个或多个屏障(barrier points)上进行协调,可以把Phaser
想象成一个多线程聚会的组织者,它负责确保所有参与的线程都到达某个阶段后再一起进行下一步。
举一个实际生活中的场景:假设正在开发一个在线多人游戏,比如,团队解谜游戏,在这个游戏中,有几个玩家(线程)需要合作完成一系列任务来通关,每个任务都被划分为几个阶段,而每个阶段都需要所有玩家共同完成某些操作后才能进入下一阶段。
这个场景中,Phaser
就可以发挥它的作用,可以把每个阶段看作是一个屏障点,每个玩家线程在完成自己当前阶段的任务后会向Phaser
报告,然后等待其他玩家完成,一旦所有玩家都完成了当前阶段的任务,Phaser
就会像一个响铃一样,通知所有玩家可以进入下一阶段了。
比如,在解谜游戏的关卡中,四个玩家需要分别找到四个不同的线索,并将这些线索组合起来才能打开通往下一关的大门,每个玩家在找到线索后,都会告知Phaser
自己已经完成任务,Phaser
会等待所有四个玩家都找到线索后,再通知他们可以将线索组合起来打开大门进入下一关了。
Phaser主要用于解决多个线程分阶段共同完成任务的同步问题,它可以确保一组线程在达到某个屏障点(phase)之前都保持同步,即所有线程都完成了某个阶段的任务后,才能一起进入下一个阶段,这种同步机制尤其适用于需要多个线程协作完成复杂任务的情况,比如在线多人游戏、分布式系统、并行计算等场景。
Phaser在内部维护了一个状态机,用来跟踪和管理每个线程的执行状态以及各个阶段的完成情况,当一个线程完成任务并达到屏障点时,会调用Phaser的相应方法来通知其他线程,然后等待其他线程一起进入下一阶段,在这个过程中,Phaser会管理线程的同步和协作,确保所有线程都能按照预定的顺序完成各自的任务。
此外,Phaser还提供了灵活的注册和注销线程的功能,可以动态地添加或删除参与同步的线程,它还支持中断和超时机制,可以在等待其他线程的过程中被中断或设置超时,增强了对多线程同步的灵活性。
官方文档:https:///jdk11/api/java.base/java/util/concurrent/Phaser.html
代码案例
下面是一个简单的Java示例代码,演示了如何使用Phaser
来同步多个线程,以确保它们分阶段完成任务,如下代码:
import java.util.concurrent.Phaser;
public class PhaserExample {
public static void main(String[] args) throws InterruptedException {
// 创建一个Phaser实例,初始时注册3个线程(不包括主线程)
Phaser phaser = new Phaser(3);
// 创建并启动3个线程
for (int i = 0; i < 3; i++) {
int threadNum = i + 1; // 为了在输出中区分线程
new Thread(() -> {
System.out.println("线程" + threadNum + ":已经准备好,等待其他线程。");
// 线程在此等待,直到所有线程都到达这个屏障点
phaser.arriveAndAwaitAdvance();
System.out.println("线程" + threadNum + ":第一阶段任务完成。");
// 模拟第二阶段的任务
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
// 再次到达屏障点,等待其他线程
phaser.arriveAndAwaitAdvance();
System.out.println("线程" + threadNum + ":第二阶段任务完成。");
// 模拟第三阶段的任务
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
// 最后一次到达屏障点,所有线程都完成后Phaser将自动进入终止状态
phaser.arriveAndAwaitAdvance();
System.out.println("线程" + threadNum + ":第三阶段任务完成,Phaser任务结束。");
}).start();
}
// 等待所有线程完成任务
// 注意:在实际应用中,可能不希望主线程在这里阻塞,而是去做其他工作
// 但为了演示目的,让主线程等待所有工作线程完成
phaser.awaitAdvance(phaser.getPhase());
System.out.println("所有线程的第一阶段任务完成。");
phaser.awaitAdvance(phaser.getPhase() + 1);
System.out.println("所有线程的第二阶段任务完成。");
phaser.awaitAdvance(phaser.getPhase() + 1);
System.out.println("所有线程的第三阶段任务完成,整个任务结束。");
}
}
在上面代码中,创建了一个Phaser
实例并初始注册了3个线程以及主线程,每个线程都执行三个阶段的任务,每个阶段任务之间都通过phaser.arriveAndAwaitAdvance()
方法进行同步,每个线程在完成当前阶段的任务后,都会在这个方法上阻塞,直到所有其他线程也完成了它们当前阶段的任务,在所有线程都完成最后一个阶段的任务后,Phaser
会自动进入终止状态,此时不会再有线程被阻塞。
上述代码输出如下结果:
主线程也已经准备好,等待其他线程。
线程x已经准备好,等待其他线程。
线程y已经准备好,等待其他线程。
线程z已经准备好,等待其他线程。
(这里所有线程和主线程都会等待,直到所有参与者都调用了arriveAndAwaitAdvance)
主线程第一阶段任务完成(实际上主线程可能只是监控或协调其他线程)。
线程x第一阶段任务完成。
线程y第一阶段任务完成。
线程z第一阶段任务完成。
(所有线程和主线程继续执行,直到它们再次调用arriveAndAwaitAdvance)
主线程第二阶段任务完成(实际上可能是等待其他线程完成某些任务)。
线程x第二阶段任务完成。
线程y第二阶段任务完成。
线程z第二阶段任务完成。
(所有线程和主线程继续执行第三阶段任务)
主线程第三阶段任务完成(实际上可能是进行一些清理工作或者汇总结果)。
线程x第三阶段任务完成,Phaser任务结束。
线程y第三阶段任务完成,Phaser任务结束。
线程z第三阶段任务完成,Phaser任务结束。
核心API
Phaser
它允许一组线程互相等待,直到所有线程都到达某个屏障(barrier)点,Phaser
非常适合用于多阶段的任务拆分和同步,以下是Phaser
中一些重要方法的简要说明:
Phaser(int parties)
: 构造函数,创建一个新的Phaser
实例,并设置注册的线程数(parties),这个数字表示在继续到下一个阶段之前,必须到达屏障的线程数。Phaser()
: 构造函数,创建一个新的Phaser
实例,但不设置注册的线程数,这通常用于层次结构的Phaser
,其中子Phaser
会继承父Phaser
的注册线程数。register()
: 增加一个到达屏障所需的线程数,如果调用此方法的线程尚未注册,它也会将自己注册为未到达的线程。arrive()
: 表示当前线程已经到达屏障,并减少未到达的线程数,如果这是最后一个到达的线程,并且已经设置了下一个阶段的屏障,那么这个方法将返回true
,否则返回false
。arriveAndAwaitAdvance()
: 当前线程到达屏障,并等待其他线程也到达,当所有线程都到达后,屏障会自动推进到下一个阶段,然后该方法返回,如果当前Phaser
被终止,这个方法会抛出IllegalStateException
。awaitAdvance(int phase)
: 等待直到屏障推进到给定的阶段,如果当前阶段大于或等于给定的阶段,那么此方法将立即返回。isTerminated()
: 检查Phaser
是否已经终止,当注册的线程数减少到零,且没有新的线程注册时,Phaser
将被终止。getPhase()
: 获取当前屏障的阶段号,每个屏障都有一个唯一的阶段号,初始阶段号为0。getRegisteredParties()
: 获取当前注册的线程数。getArrivedParties()
: 获取已经到达当前屏障的线程数。getUnarrivedParties()
: 获取尚未到达当前屏障的线程数,这实际上是getRegisteredParties()
和getArrivedParties()
之间的差值。forceTermination()
: 强制终止Phaser
,即使还有未到达的线程,这会导致所有等待在arriveAndAwaitAdvance()
或awaitAdvance(int)
方法上的线程抛出IllegalStateException
。onAdvance(int phase, int registeredParties)
: 这是一个受保护的方法,可以在子类中覆盖,以便在每个屏障阶段推进时执行自定义操作。bulkRegister(int parties)
: 一次性注册多个线程,这通常用于静态已知的线程数,或者当多个任务由同一个线程代表时。
核心总结
Phaser类的目的是允许在并发编程中同步多个线程之间的执行,它具有如下优点,如下:
- 更好的可扩展性:Phaser类相对于其他同步工具类(如CyclicBarrier和CountDownLatch)具有更好的可扩展性,因为它支持更多的参与者(即线程)同时进行同步。
- 自动注销和清理:当所有参与者都完成执行后,Phaser会自动注销并释放相关资源,这有助于避免内存泄漏和资源浪费。
- 灵活的执行模式:Phaser类提供了多种执行模式,如并行、串行和混合模式,这使得在处理并发任务时更加灵活。
它也有不少缺点,如:1、与其他的同步工具类相比,Phaser类的实现相对复杂,因此在某些场景下可能会引入额外的性能开销,并且Phaser类具有一定的使用门槛,使用时深入理解并发编程和Java并发API,这可能会增加学习成本。