黑马程序员技术交流社区

标题: 【上海校区】Java并发——Phaser “阶段器” [打印本页]

作者: 不二晨 时间: 2018-11-27 09:17
标题: 【上海校区】Java并发——Phaser “阶段器”
1. Phaser

Phaser是一个更加弹性的同步屏障。类java.util.concurrent实现了Phaser.

Phaser表示“阶段器”，用来解决控制多个线程分阶段共同完成任务的情景问题。其作用相比CountDownLatch和CyclicBarrier更加灵活，例如有这样的一个题目：5个学生一起参加考试，一共有三道题，要求所有学生到齐才能开始考试，全部同学都做完第一题，学生才能继续做第二题，全部学生做完了第二题，才能做第三题，所有学生都做完的第三题，考试才结束。分析这个题目：这是一个多线程（5个学生）分阶段问题（考试考试、第一题做完、第二题做完、第三题做完），所以很适合用Phaser解决这个问题。

2. 常用方法

（1）Phaser(int parties)构造函数

与CountDownLatch一样，传入同步的线程数，也支持层次构造Phaser(Phaser parent)。

（2）register()

动态添加一个或多个参与者，同时返回phase值作抵达分类用。

（3）bulkRegister(int parties)

将指定数目的参与者注册到phaser中，所有这些新的参与者都将被当成没有执行完本阶段的线程。

（4）int arriveAndAwaitAdvance()

类似await()方法，记录到达线程数，阻塞等待其他线程到达同步点后再继续执行。

（5）arriveAndDeregister()

动态撤销线程在phaser的注册，通知phaser对象，该线程已经结束该阶段且不参与后面阶段。由此减少了未来phase上需要前进的线程数量。

（6）arrive()

通知phaser该线程已经完成该阶段，但不等待其他线程。必须小心使用这个方法，因为它不会与其他线程同步。

（7）forceTermination()

强制phaser进入终止状态，不管是否存在未注册的参与线程，当一个线程出现错误时，强制终止phaser是很有意义的。

（8）boolean onAdvance(int phase, int registeredParties)方法。此方法有2个作用：
      当每一个阶段执行完毕，此方法会被自动调用，因此，重载此方法写入的代码会在每个阶段执行完毕时执行，相当于CyclicBarrier的barrierAction。
      当此方法返回true时，意味着Phaser被终止，因此可以巧妙的设置此方法的返回值来终止所有线程。

（9）在Phaser内有2个重要状态，分别是phase和party。
      phase就是阶段，初值为0，当所有的线程执行完本轮任务，同时开始下一轮任务时，意味着当前阶段已结束，进入到下一阶段，phase的值自动加1。

      party就是线程， party=4就意味着Phaser对象当前管理着4个线程。

phaser有一个重大特性，就是不必对它的方法进行异常处理。置于休眠的线程不会响应中断事件，不会抛出interruptedException异常，只有一个方法会响应：AwaitAdvanceInterruptibly(int phaser).

3.示例

5个学生一起参加考试，一共有三道题，要求所有学生到齐才能开始考试，全部同学都做完第一题，学生才能继续做第二题，全部学生做完了第二题，才能做第三题，所有学生都做完的第三题，考试才结束。分析这个题目：这是一个多线程（5个学生）分阶段问题（考试考试、第一题做完、第二题做完、第三题做完），所以很适合用Phaser解决这个问题。

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

public class PhaserDemo {

public static void main(String[] args)
{
MyPhaser phaser = new MyPhaser();
StudentTask[] tasks = new StudentTask[5];
for (int i = 0; i < tasks.length; i++) {
tasks = new StudentTask(phaser);
phaser.register(); //注册一次表示phaser维护的线程个数
}
Thread[] threads = new Thread[tasks.length];
for (int i = 0; i < tasks.length; i++) {
threads = new Thread(tasks, "Student "+i);
threads.start();
}

//等待所有线程执行结束
for (int i = 0; i < tasks.length; i++) {
try {
threads.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

System.out.println("Phaser has finished:"+phaser.isTerminated());

}
}

class StudentTask implements Runnable{
private Phaser phaser;

public StudentTask(Phaser phaser) {
this.phaser = phaser;
}

Random random = new Random();
@Override
public void run()
{
System.out.println(Thread.currentThread().getName()+"已做好准备");
phaser.arriveAndAwaitAdvance();

System.out.println(Thread.currentThread().getName()+"开始做第1题");
int time1 = doExercise1();
System.out.println(Thread.currentThread().getName()+"完成第1题，做题时间：" + time1);
phaser.arriveAndAwaitAdvance();

System.out.println(Thread.currentThread().getName()+"开始做第2题");
int time2 = doExercise2();
System.out.println(Thread.currentThread().getName()+"完成第2题，做题时间：" + time2);
phaser.arriveAndAwaitAdvance();

System.out.println(Thread.currentThread().getName()+"开始做第3题");
int time3 = doExercise3();
System.out.println(Thread.currentThread().getName()+"完成第3题，做题时间：" + time3);
phaser.arriveAndAwaitAdvance();

}

private int doExercise1() {
int time1 = random.nextInt(100);
try {
TimeUnit.SECONDS.sleep(time1);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return time1;
}

private int doExercise2() {
int time2 = random.nextInt(100);
try {
TimeUnit.SECONDS.sleep(time2);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return time2;
}

private int doExercise3() {
int time3 = random.nextInt(100);
try {
TimeUnit.SECONDS.sleep(time3);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return time3;
}

}

class MyPhaser extends Phaser{
   @Override
   protected boolean onAdvance(int phase, int registeredParties)
   {
   switch(phase) {
      case 0:
           return studentArrived();
      case 1:
           return finishFirstExercise();
      case 2:
           return finishSecondExercise();
      case 3:
           return finishExam();
      default:
           return true;
   }
   }

   private boolean studentArrived() {
   System.out.println("全部考生都已做好准备，考生人数：" + getRegisteredParties());
   return false;
}

   private boolean finishFirstExercise() {
System.out.println("所有考生都已完成第1题");
return false;
}

   private boolean finishSecondExercise() {
System.out.println("所有考生都已完成第2题");
return false;
}

   private boolean finishExam() {
System.out.println("所有考生都已完成第3题，结束考试！");
return true;
}
}
运行结果：

Student 1已做好准备
Student 4已做好准备
Student 2已做好准备
Student 3已做好准备
Student 0已做好准备
全部考生都已做好准备，考生人数：5
Student 0开始做第1题
Student 4开始做第1题
Student 3开始做第1题
Student 1开始做第1题
Student 2开始做第1题
Student 2完成第1题，做题时间：32
Student 0完成第1题，做题时间：59
Student 3完成第1题，做题时间：60
Student 1完成第1题，做题时间：69
Student 4完成第1题，做题时间：96
所有考生都已完成第1题
Student 0开始做第2题
Student 1开始做第2题
Student 4开始做第2题
Student 2开始做第2题
Student 3开始做第2题
Student 2完成第2题，做题时间：8
Student 0完成第2题，做题时间：38
Student 3完成第2题，做题时间：69
Student 1完成第2题，做题时间：73
Student 4完成第2题，做题时间：87
所有考生都已完成第2题
Student 1开始做第3题
Student 3开始做第3题
Student 2开始做第3题
Student 0开始做第3题
Student 4开始做第3题
Student 2完成第3题，做题时间：23
Student 1完成第3题，做题时间：25
Student 0完成第3题，做题时间：25
Student 4完成第3题，做题时间：37
Student 3完成第3题，做题时间：57
所有考生都已完成第3题，结束考试！
Phaser has finished:true
---------------------
【转载】
作者：zhm1563550235
原文：https://blog.csdn.net/zhm1563550235/article/details/84453906

作者: 不二晨 时间: 2018-11-28 15:40
奈斯

欢迎光临黑马程序员技术交流社区 (http://bbs.itheima.com/) 黑马程序员IT技术论坛 X3.2