并发编程系列之Phaser用法简介

Posted smileNicky

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了并发编程系列之Phaser用法简介相关的知识,希望对你有一定的参考价值。

1、什么是相位器Phaser

jdk7中增加了一个用于多阶段同步控制的工具类,它包含了CyclicBarrier和CountDownLatch的相关功能,比它们更强大灵活。

对Phaser阶段协同器的理解,Phaser适用于多个线程协作的任务,分为多个阶段,每个阶段都可以有任意个参与者,线程可以随时注册并参与某个阶段;当一个阶段中所有任务都成功完成后,Phaser的onAdvance()被调用,然后Phaser释放等待线程,自动进入下个阶段。如此循环,直到Phaser不再包含任何参与者。

2、Phaser API说明

  • 构造方法

    • Phaser() :参与任务数0
    • Phaser(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();
				
			

		

	


以上是关于并发编程系列之Phaser用法简介的主要内容,如果未能解决你的问题,请参考以下文章

并发编程系列之ReentrantLock用法简介

并发编程系列之ReentrantLock用法简介

并发编程系列之Semaphore用法简介

并发编程系列之Semaphore用法简介

并发编程系列之CountDownLatch用法简介

并发编程系列之CountDownLatch用法简介