并发编程系列之Phaser用法简介
2023-03-20 14:55:20 时间
1、什么是相位器Phaser
jdk7中增加了一个用于多阶段同步控制的工具类,它包含了CyclicBarrier和CountDownLatch的相关功能,比它们更强大灵活。
对Phaser阶段协同器的理解,Phaser适用于多个线程协作的任务,分为多个阶段,每个阶段都可以有任意个参与者,线程可以随时注册并参与某个阶段;当一个阶段中所有任务都成功完成后,Phaser的onAdvance()被调用,然后Phaser释放等待线程,自动进入下个阶段。如此循环,直到Phaser不再包含任何参与者。
2、Phaser API说明
- 构造方法
Phaser()
:参与任务数0Phaser(int parties)
:指定初始参与任务数Phaser(Phaser parent)
:指定parent阶段器, 子对象作为一个整体加入parent对象,当子对象中没有参与者时,会自动从parent对象解除注册Phaser(Phaser parent , int parties)
:集成上面两个方法的
- 增减参与任务数方法
int register()
:增加一个数,返回当前阶段号int bulkRegister(int parties)
:增加指定个数,返回当前阶段号int arriveAndDeregister()
:减少一个任务数,返回当前阶段号
- 到达等待方法
int arrive()
:到达,任务完成,返回当前阶段号int arriveAndAwaitAdvance()
:到达后等待其他任务到达,返回到达阶段号int awaitAdvance(int phase)
:在指定阶段等待(必须是当前阶段才有效)int awaitAdvanceInterruptibly(int phase)
int awaitAdvanceInterruptibly(int phase , long timeout, TimeUnit unit)
- 阶段到达触发动作
protected boolean onAdvance(int Phase , int registeredParties)
:类似于CyclicBarrier的触发命令,通过重写该方法来增加阶段到达动作
- 其它api
void forceTermination()
:强制结束boolean isTerMinated()
:判断是否结束void getPhase()
:获取当前阶段号
- 注意事项 : 单个Phaser实例允许的注册任务数的上限是65535,如果参与任务数超过,可以用父子Phaser树的方式
3、Phaser原理图
4、Phaser例子
场景:公司组织郊游活动,大家各自从家出发到公司集合,大家都到了后,出发到公园各自游玩,然后在公园门口集合,再去餐厅就餐,大家都到了就开始用餐。有的员工白天有事,选择晚上的聚餐,有的员工则晚上有事,只参加白天的活动。编程模拟实现。
- 第一阶段,到公司集合5人,任务数为5,去公园游玩。
- 第二阶段,到公园门口集合,有2人因为晚上有事,自行从公园回家;则3人去餐厅,这是减少参与数,任务数变为3
- 第三阶段,餐厅集合,有另外4人参加聚餐,这是增加参与数,任务数变为7
import java.util.Random;
import java.util.concurrent.Phaser;
public class MultipleStartTogetherPhserDemo {
Random rd = new Random();
int bound = 5000;
public void step1Task() throws InterruptedException {
// 经过一段时间后,到达公司
Thread.sleep(rd.nextInt(bound));
System.out.println(
"员工【" + Thread.currentThread().getName() + "】" + "到达公司!");
}
public void step2Task() throws InterruptedException {
System.out.println(
"员工【" + Thread.currentThread().getName() + "】" + "出发去公园玩...");
// 玩了一段时间后,到公园门口集合
Thread.sleep(rd.nextInt(bound));
System.out.println(
"员工【" + Thread.currentThread().getName() + "】" + "完成公园游玩!");
}
public void step3Task() throws InterruptedException {
System.out.println(
"员工【" + Thread.currentThread().getName() + "】" + "出发去餐厅...");
// 玩了一段时间后,到公园门口集合
Thread.sleep(rd.nextInt(bound));
System.out.println(
"员工【" + Thread.currentThread().getName() + "】" + "到达餐厅!");
}
public void step4Task() throws InterruptedException {
System.out.println(
"员工【" + Thread.currentThread().getName() + "】" + "开始用餐...");
// 玩了一段时间后,到公园门口集合
Thread.sleep(rd.nextInt(bound));
System.out.println(
"员工【" + Thread.currentThread().getName() + "】" + "回家了!");
}
public static void main(String[] args) {
// 创建阶段协同器对象,重写了onAdvance方法,增加阶段到达处理逻辑
final Phaser ph = new Phaser() {
protected boolean onAdvance(int phase, int registeredParties) {
int staffs = registeredParties - 1;
switch (phase) {
case 0:
System.out.println("大家都到公司了,出发去公园!人数:" + staffs);
break;
case 1:
System.out.println("大家都到公园大门,出发去餐厅!人数:" + staffs);
break;
case 2:
System.out.println("大家都到餐厅了,开始用餐!人数:" + staffs);
break;
}
// 判断是否只剩主线程一个参与者,是,则返回true,阶段协同器终止。
return registeredParties == 1;
}
};
// 增加一个任务数,用来让主线程全程参与
ph.register();
final MultipleStartTogetherPhserDemo job = new MultipleStartTogetherPhserDemo();
// 让3个全程参与的线程加入
for (int i = 0; i < 3; i++) {
// 增加参与任务数
ph.register();
new Thread(new Runnable() {
@Override
public void run() {
try {
job.step1Task();
ph.arriveAndAwaitAdvance();
job.step2Task();
System.out.println(
"员工【" + Thread.currentThread().getName() + "】"
+ "到达公园大门集合");
ph.arriveAndAwaitAdvance();
job.step3Task();
ph.arriveAndAwaitAdvance();
job.step4Task();
// 完成了,注销离开
ph.arriveAndDeregister();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
// 让两个不参加聚餐的员工加入
for (int i = 0; i < 2; i++) {
// 增加参与任务数
ph.register();
new Thread(new Runnable() {
@Override
public void run() {
try {
job.step1Task();
ph.arriveAndAwaitAdvance();
job.step2Task();
System.out.println(
"员工【" + Thread.currentThread().getName() + "】"
+ "回家了!");
// 完成了,注销离开
ph.arriveAndDeregister();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
while (!ph.isTerminated()) {
int phaser = ph.arriveAndAwaitAdvance();
if (phaser == 2) { // 到了去餐厅的阶段,让只参加晚上聚餐的人加入
for (int i = 0; i < 4; i++) {
// 增加参与任务数
ph.register();
new Thread(new Runnable() {
@Override
public void run() {
try {
job.step3Task();
ph.arriveAndAwaitAdvance();
job.step4Task();
// 完成了,注销离开
ph.arriveAndDeregister();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
}
}
}
相关文章
- 深入探讨Java中的异常与错误处理
- 研究学习Kotlin的一些方法
- 数据显示Java热度持续下落,日子屈指可数?
- 2017年5月编程语言排行榜:Java与C语言优势正开始缩小
- Java多线程之内置锁与显示锁
- Java线程池的理论与实践
- 白话阿里巴巴Java开发手册(编程规约)
- 关于Java你不知道的十件事
- Java服务化系统线上应急和技术攻关,你必须掌握的Linux命令
- Java实现高斯模糊和图像的空间卷积
- Java阻塞队列实现原理分析
- NPM使用技巧
- Node.js对Java开发者而言是什么?
- Java反射机制应用实践
- 理解RxJava中的Single和Completable
- 2017年你不能错过的Java类库
- 大规模集群下的Hadoop NameNode
- 从源码解密Spark内存管理
- 2017年3月编程语言排行榜:Swift首次进入前十
- JVM热点技术:Java类的加载机制