并发编程系列之Phaser用法简介
【摘要】
1、什么是相位器Phaser
jdk7中增加了一个用于多阶段同步控制的工具类，它包含了CyclicBarrier和CountDownLatch的相关功能，比它们更强大灵活。
对Phaser阶段协同器的...
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();
}
}
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
文章来源: smilenicky.blog.csdn.net，作者：smileNicky，版权归原作者所有，如需转载，请联系作者。
原文链接：smilenicky.blog.csdn.net/article/details/121612813
【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）