高级同步器:可重用的同步屏障Phaser

引自:https://shift-alt-ctrl.iteye.com/blog/2302923

在JAVA 1.7引入了一个新的并发API:Phaser,一个可重用的同步barrier。在此前,JAVA已经有CyclicBarrier、CountDownLatch这两种同步barrier,但是Phaser更加灵活,而且侧重于“重用”。

一、简述

1、注册机制:与其他barrier不同的是,Phaser中的“注册的同步者(parties)”会随时间而变化,Phaser可以通过构造器初始化parties个数,也可以在Phaser运行期间随时加入(register)新的parties,以及在运行期间注销(deregister)parties。运行时可以随时加入、注销parties,只会影响Phaser内部的计数器,它建立任何内部的bookkeeping(账本),因此task不能查询自己是否已经注册了,当然你可以通过实现子类来达成这一设计要求。

Java代码  

  1. //伪代码
  2. Phaser phaser = new Phaser();
  3. phaser.register();//parties count: 1
  4. ....
  5. phaser.arriveAndDeregister()://count : 0;
  6. ....

此外,CyclicBarrier、CountDownLatch需要在初始化的构造函数中指定同步者的个数,且运行时无法再次调整。

Java代码  

  1. CountDownLatch countDownLatch = new CountDownLatch(12);
  2. //count deregister parties after all
  3. //parties count is 12 all the times
  4. //if you want change the number of parties, you should create a new instance.
  5. CyclicBarrier cyclicBarrier = new CyclicBarrier(12);

2、同步机制:类似于CyclicBarrier,Phaser也可以awaited多次,它的arrivedAndAwaitAdvance()方法的效果类似于CyclicBarrier的await()。Phaser的每个周期(generation)都有一个phase数字,phase 从0开始,当所有的已注册的parties都到达后(arrive)将会导致此phase数字自增(advance),当达到Integer.MAX_VALUE后继续从0开始。这个phase数字用于表示当前parties所处于的“阶段周期”,它既可以标记和控制parties的wait行为、唤醒等待的时机。

1)Arrival:Phaser中的arrive()、arriveAndDeregister()方法,这两个方法不会阻塞(block),但是会返回相应的phase数字,当此phase中最后一个party也arrive以后,phase数字将会增加,即phase进入下一个周期,同时触发(onAdvance)那些阻塞在上一phase的线程。这一点类似于CyclicBarrier的barrier到达机制;更灵活的是,我们可以通过重写onAdvance方法来实现更多的触发行为。

2)Waiting:Phaser中的awaitAdvance()方法,需要指定一个phase数字,表示此Thread阻塞直到phase推进到此周期,arriveAndAwaitAdvance()方法阻塞到下一周期开始(或者当前phase结束)。不像CyclicBarrier,即使等待Thread已经interrupted,awaitAdvance方法会继续等待。Phaser提供了Interruptible和Timout的阻塞机制,不过当线程Interrupted或者timout之后将会抛出异常,而不会修改Phaser的内部状态。如果必要的话,你可以在遇到此类异常时,进行相应的恢复操作,通常是在调用forceTermination()方法之后。

Phaser通常在ForJoinPool中执行tasks,它可以在有task阻塞等待advance时,确保其他tasks的充分并行能力。

3、中断(终止):Phaser可以进入Termination状态,可以通过isTermination()方法判断;当Phaser被终止后,所有的同步方法将会立即返回(解除阻塞),不需要等到advance(即advance也会解除阻塞),且这些阻塞方法将会返回一个负值的phase值(awaitAdvance方法、arriveAndAwaitAdvance方法)。当然,向一个termination状态的Phaser注册party将不会有效;此时onAdvance()方法也将会返回true(默认实现),即所有的parties都会被deregister,即register个数为0。

4、Tiering(分层):Phaser可以“分层”,以tree的方式构建Phaser来降低“竞争”。如果一个Phaser中有大量parties,这会导致严重的同步竞争,所以我们可以将它们分组并共享一个parent Phaser,这样可以提高吞吐能力;Phaser中注册和注销parties都会有Child 和parent Phaser自动管理。当Child Phaser中中注册的parties变为非0时(在构造函数Phaser(Phaser parent,int parties),或者register()方法),Child Phaser将会注册到其Parent上;当Child Phaser中的parties变为0时(比如由arrivedAndDegister()方法),那么此时Child Phaser也将从其parent中注销出去。

5、监控:同步的方法只会被register操作调用,对于当前state的监控方法可以在任何时候调用,比如getRegisteredParties()获取已经注册的parties个数,getPhase()获取当前phase周期数等;因为这些方法并非同步,所以只能反映当时的瞬间状态。

二、常用的Barrier比较

1、CountDownLatch

Java代码  

//创建时,就需要指定参与的parties个数
int parties = 12;
CountDownLatch latch = new CountDownLatch(parties);
//线程池中同步task
ExecutorService executor = Executors.newFixedThreadPool(parties);
for(int i = 0; i < parties; i++) {
    executor.execute(new Runnable() {
        @Override
        public void run() {
            try {
                //可以在任务执行开始时执行,表示所有的任务都启动后,主线程的await即可解除
                //latch.countDown();
                //run
                //..
                Thread.sleep(3000);  

            } catch (Exception e) {  

            }
            finally {
                //任务执行完毕后:到达
                //表示所有的任务都结束,主线程才能继续
                latch.countDown();
            }
        }
    });
}
latch.await();//主线程阻塞,直到所有的parties到达
//latch上所有的parties都达到后,再次执行await将不会有效,
//即barrier是不可重用的
executor.shutdown();  

2、CyclicBarrier

Java代码  

//创建时,就需要指定参与的parties个数
int parties = 12;
CyclicBarrier barrier = new CyclicBarrier(parties);
//线程池中同步task
ExecutorService executor = Executors.newFixedThreadPool(parties);
for(int i = 0; i < parties; i++) {
    executor.execute(new Runnable() {
        @Override
        public void run() {
            try {
                int i = 0;
                while (i < 3 && !barrier.isBroken()) {
                    System.out.println("generation begin:" + i + ",tid:" + Thread.currentThread().getId());
                    Thread.sleep(3000);
                    //如果所有的parties都到达,则开启新的一次周期(generation)
                    //barrier可以被重用
                    barrier.await();
                    i++;
                }  

            } catch (Exception e) {
                e.printStackTrace();
            }
            finally {  

            }
        }
    });
}
Thread.sleep(100000);  

3、Phaser

Java代码  

package com.thread;

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

/**
 * 同步器:Phaser,可重用的同步屏障
 * @author Administrator
 *
 */
public class PhaserTest {

    public static void main(String[] args) {

        // 创建时,就需要指定参与的parties个数
        // 可以在创建时不指定parties
        // 而是在运行时,随时注册和注销新的parties
        int parties = 12;

        Phaser phaser = new Phaser();
        // 主线程先注册一个
        // 对应下文中,主线程可以等待所有的parties到达后再解除阻塞(类似与CountDownLatch)
        phaser.register();
        ExecutorService executor = Executors.newFixedThreadPool(parties);
        for (int i = 0; i < parties; i++) {
            phaser.register();// 每创建一个task,我们就注册一个party
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        int i = 0;
                        while (i < 3 && !phaser.isTerminated()) {
                            //getPhase()获取当前phase周期数
                            System.out.println("Generation:" + phaser.getPhase());
                            Thread.sleep(3000);
                            // 等待同一周期内,其他Task到达
                            // 然后进入新的周期,并继续同步进行
                            phaser.arriveAndAwaitAdvance();
                            i++;// 我们假定,运行三个周期即可
                        }
                    } catch (Exception e) {

                    } finally {
                        phaser.arriveAndDeregister();
                    }
                }
            });
        }
        // 主线程到达,且注销自己
        // 此后线程池中的线程即可开始按照周期,同步执行。
        phaser.arriveAndDeregister();
        executor.shutdown();
    }
}

三、API简述

1、Phaser():构造函数,创建一个Phaser;默认parties个数为0。此后我们可以通过register()、bulkRegister()方法来注册新的parties。每个Phaser实例内部,都持有几个状态数据:termination状态、已经注册的parties个数(registeredParties)、当前phase下已到达的parties个数(arrivedParties)、当前phase周期数,还有2个同步阻塞队列Queue。Queue中保存了所有的waiter,即因为advance而等待的线程信息;这两个Queue分别为evenQ和oddQ,这两个Queue在实现上没有任何区别,Queue的元素为QNode,每个QNode保存一个waiter的信息,比如Thread引用、阻塞的phase、超时的deadline、是否支持interrupted响应等。两个Queue,其中一个保存当前phase中正在使用的waiter,另一个备用,当phase为奇数时使用evenQ、oddQ备用,偶数时相反,即两个Queue轮换使用。当advance事件触发期间,新register的parties将会被放在备用的Queue中,advance只需要响应另一个Queue中的waiters即可,避免出现混乱。

2、Phaser(int parties):构造函数,初始一定数量的parties;相当于直接regsiter此数量的parties。

3、arrive():到达,阻塞,等到当前phase下其他parties到达。如果没有register(即已register数量为0),调用此方法将会抛出异常,此方法返回当前phase周期数,如果Phaser已经终止,则返回负数。

4、arriveAndDeregister():到达,并注销一个parties数量,非阻塞方法。注销,将会导致Phaser内部的parties个数减一(只影响当前phase),即下一个phase需要等待arrive的parties数量将减一。异常机制和返回值,与arrive方法一致。

5、arriveAndAwaitAdvance():到达,且阻塞直到其他parties都到达,且advance。此方法等同于awaitAdvance(arrive())。如果你希望阻塞机制支持timeout、interrupted响应,可以使用类似的其他方法(参见下文)。如果你希望到达后且注销,而且阻塞等到当前phase下其他的parties到达,可以使用awaitAdvance(arriveAndDeregister())方法组合。此方法的异常机制和返回值同arrive()。

6、awaitAdvance(int phase):阻塞方法,等待phase周期数下其他所有的parties都到达。如果指定的phase与Phaser当前的phase不一致,则立即返回。

7、awaitAdvanceInterruptibly(int phase):阻塞方法,同awaitAdvance,只是支持interrupted响应,即waiter线程如果被外部中断,则此方法立即返回,并抛出InterrutedException。

8、awaitAdvanceInterruptibly(int phase,long timeout,TimeUnit unit):阻塞方法,同awaitAdvance,支持timeout类型的interrupted响应,即当前线程阻塞等待约定的时长,超时后以TimeoutException异常方式返回。

9、forceTermination():强制终止,此后Phaser对象将不可用,即register等将不再有效。此方法将会导致Queue中所有的waiter线程被唤醒。

10、register():新注册一个party,导致Phaser内部registerPaties数量加1;如果此时onAdvance方法正在执行,此方法将会等待它执行完毕后才会返回。此方法返回当前的phase周期数,如果Phaser已经中断,将会返回负数。

11、bulkRegister(int parties):批量注册多个parties数组,规则同10、。

12、getArrivedParties():获取已经到达的parties个数。

13、getPhase():获取当前phase周期数。如果Phaser已经中断,则返回负值。

14、getRegisteredParties():获取已经注册的parties个数。

15、getUnarrivedParties():获取尚未到达的parties个数。

16、onAdvance(int phase,int registeredParties):这个方法比较特殊,表示当进入下一个phase时可以进行的事件处理,如果返回true表示此Phaser应该终止(此后将会把Phaser的状态为termination,即isTermination()将返回true。),否则可以继续进行。phase参数表示当前周期数,registeredParties表示当前已经注册的parties个数。

默认实现为:return registeredParties == 0;在很多情况下,开发者可以通过重写此方法,来实现自定义的advance时间处理机制。

内部原理,比较简单(简述):

1)两个计数器:分别表示parties个数和当前phase。register和deregister会触发parties变更(CAS),全部parties到达(arrive)会触发phase变更。

2)一个主要的阻塞队列:非AQS实现,对于arriveAndWait的线程,会被添加到队列中并被park阻塞,知道当前phase中最后一个party到达后触发唤醒。

原文地址:https://www.cnblogs.com/x-jingxin/p/10655164.html

时间: 2024-10-04 22:12:10

高级同步器:可重用的同步屏障Phaser的相关文章

concurrent(四)同步屏障 CyclicBarrier &amp; 源码分析

参考文档:Java多线程系列--"JUC锁"10之 CyclicBarrier原理和示例:https://www.cnblogs.com/skywang12345/p/3533995.html简介CyclicBarrier是一个同步辅助类,允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point).因为该 barrier 在释放等待线程后可以重用,所以称它为循环的 barrier.基于ReentrantLock实现举个栗子 /** * 简单模拟一下对战平

倒计数锁存器(CountDown Latch)和 CyclicBarrier(同步屏障)

倒计数锁存器(CountDown Latch)是异常性障碍,允许一个或多个线程等待一个或者多个其他线程来做某些事情. public static long time(Executor executor,int concurrency,final Runnable action)throws InterruptedException{ //CountDownLatch 构造器参数代表 只有调用countDown()方法达到这个int值的时候,才能继续(所以多线程的情况下可以在此阻塞) final

Java并发工具类(二)同步屏障CyclicBarrier

作用 CyclicBarrier是一个同步辅助类,它允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point),才继续执行. 简介 CyclicBarrier 的字面意思是可循环使用(Cyclic)的屏障(Barrier).它要做的事情是,让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续干活.CyclicBarrier默认的构造方法是CyclicBarrier(int parties),其参数

并发工具类(二)同步屏障CyclicBarrier

前言 ??JDK中为了处理线程之间的同步问题,除了提供锁机制之外,还提供了几个非常有用的并发工具类:CountDownLatch.CyclicBarrier.Semphore.Exchanger.Phaser: ??CountDownLatch.CyclicBarrier.Semphore.Phaser 这四个工具类提供一种并发流程的控制手段:而Exchanger工具类则提供了在线程之间交换数据的一种手段. 简介 ??CyclicBarrier 的字面意思是可循环使用(Cyclic)的屏障(Ba

java多线程--同步屏障CyclicBarrier的使用

CyclicBarrier的概念理解: CyclicBarrier的字面上的意思是可循环的屏障,是java并发包java.util.concurrent 里的一个同步工具类,在我下载的JDK1.6的中文文档里对他的解释是: 大体意思就是:让一组线程到达一个屏障,一个集合点时,被阻塞,直到所有的线程都到了这个集合点时,屏障才会打开,然后线程才能继续往下执行.举个简单的例子就是:旅游团带着一帮人参观景点,规定在下一个景点A处集合,于是导游就在景点A等着大家,导游就是这个集合点或者说屏障,直到所有的游

Linux线程同步------屏障

屏障是Linux中协调多个线程并行工作的同步机制,屏障允许每个线程等待直到所有的合作线程到达某一点,然后继续从该点执行,pthread_join是一种屏障但只允许一个线程等待,pthread_barrier允许任意数量的线程等待! pthread_barrier_init(pthread_barrier_t *屏障,属性,unsigned int 屏障计数值): pthread_barrier_wait(pthread_barrier_t *屏障):在每个线程中调用则计数值加一并将当前线程阻塞在

Java并发工具类之同步屏障CyclicBarrier

CyclicBarrier的字面意思是可以循环使用的Barrier,它要做的事情是让一个线程到达一个Barrier的时候被阻塞,直到最后一个线程到达Barrier,屏障才会放开,所有被Barrier拦截的线程才会继续运行. CyclicBarrier的默认的构造器是CyclicBarrier(int parties),参数表示屏障拦截的线程数量,每个线程调用await方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞.示例代码如下: public class CyclicBa

java并发之同步辅助类Phaser

Phaser含义: 更加复杂和强大的同步辅助类.它允许并发执行多阶段任务.当我们有并发任务并且需要分解成几步执行时,(CyclicBarrier是分成两步),就可以选择使用Phaser.Phaser类机制是在每一步结束的位置对线程进行同步,当所有的线程都完成了这一步,才允许执行下一步.跟其他同步工具一样,必须对Phaser类中参与同步操作的任务数进行初始化,不同的是,可以动态的增加或者减少任务数. 函数:arriveAndAwaitAdvance():类似于CyclicBarrier的await

java CyclicBarrier同步屏障

CyclicBarrier的字面意思是可循环使用的屏障,它的主要作用是,让一组线程到达一个屏障时被阻塞,知道最后一个线程到达屏障时,屏障才会打开,所有被屏障拦截的线程才会继续运行. 1.简介: CyclicBarrier默认的构造方法是CyclicBarrier(int parties),其中参数标识屏障拦截的线程数量,每个线程调用await方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞. package com.test; import java.util.concur