【Java并发编程实战】—–“J.U.C”:Phaser

Phaser由java7中推出,是Java SE 7中新增的一个使用同步工具,在功能上面它与CyclicBarrierCountDownLatch有些重叠,但是它提供了更加灵活、强大的用法。

CyclicBarrier,允许一组线程互相等待,直到到达某个公共屏障点。它提供的await()可以实现让所有参与者在临界点到来之前一直处于等待状态。

CountDownLatch,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。它提供了await()、countDown()两个方法来进行操作。

在Phaser中,它把多个线程协作执行的任务划分为多个阶段,编程时需要明确各个阶段的任务,每个阶段都可以有任意个参与者,线程都可以随时注册并参与到某个阶段。

构造

Phaser创建后,初始阶段编号为0,构造函数中指定初始参与个数。

注册:Registration

Phaser支持通过register()和bulkRegister(int parties)方法来动态调整注册任务的数量。

Arrival

每个Phaser实例都会维护一个phase number,初始值为0。每当所有注册的任务都到达Phaser时,phase number累加,并在超过Integer.MAX_VALUE后清零。arrive()和arriveAndDeregister()方法用于记录到达;其中arrive(),某个参与者完成任务后调用;arriveAndDeregister(),任务完成,取消自己的注册。arriveAndAwaitAdvance(),自己完成等待其他参与者完成,进入阻塞,直到Phaser成功进入下个阶段。

example 1

public class PhaserTest_1 {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(5);

        for(int i = 0 ; i < 5 ; i++){
            Task_01 task_01 = new Task_01(phaser);
            Thread thread = new Thread(task_01, "PhaseTest_" + i);
            thread.start();
        }
    }

    static class Task_01 implements Runnable{
        private final Phaser phaser;

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

        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + "执行任务完成,等待其他任务执行......");
            //等待其他任务执行完成
            phaser.arriveAndAwaitAdvance();
            System.out.println(Thread.currentThread().getName() + "继续执行任务...");
        }
    }
}

运行结果:

PhaseTest_0执行任务完成,等待其他任务执行......
PhaseTest_1执行任务完成,等待其他任务执行......
PhaseTest_3执行任务完成,等待其他任务执行......
PhaseTest_2执行任务完成,等待其他任务执行......
PhaseTest_4执行任务完成,等待其他任务执行......
PhaseTest_4继续执行任务...
PhaseTest_1继续执行任务...
PhaseTest_0继续执行任务...
PhaseTest_2继续执行任务...
PhaseTest_3继续执行任务...

在该实例中我们可以确认,所有子线程的****+”继续执行任务…”,都是在线程调用arriveAndAwaitAdvance()方法之后执行的。

example 2

前面提到过,Phaser提供了比CountDownLatch、CyclicBarrier更加强大、灵活的功能,从某种程度上来说,Phaser可以替换他们:

public class PhaserTest_5 {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(1);        //相当于CountDownLatch(1) 

        //五个子任务
        for(int i = 0 ; i < 3 ; i++){
            Task_05 task = new Task_05(phaser);
            Thread thread = new Thread(task,"PhaseTest_" + i);
            thread.start();
        }

        try {
            //等待3秒
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        phaser.arrive();        //countDownLatch.countDown()
    }

    static class Task_05 implements Runnable{
        private final Phaser phaser;

        Task_05(Phaser phaser){
            this.phaser = phaser;
        }

        @Override
        public void run() {
            phaser.awaitAdvance(phaser.getPhase());        //countDownLatch.await()
            System.out.println(Thread.currentThread().getName() + "执行任务...");
        }
    }
}

在这里,任务一开始并没有真正执行,而是等待三秒后执行。

对于CyclicBarrier就更加简单了,直接arriveAndAwaitAdvance()方法替换,如example 1。

example 3

在CyclicBarrier中当任务执行完之后可以执行一个action,在Phaser中同样有一个对应的action,只不过Phaser需要重写onAdvance()方法:

public class PhaserTest_3 {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(3){
            /**
             * registeredParties:线程注册的数量
             * phase:进入该方法的线程数,
             */
             protected boolean onAdvance(int phase, int registeredParties) {
                 System.out.println("执行onAdvance方法.....;phase:" + phase + "registeredParties=" + registeredParties);
                 return phase == 3;
             }
        };

        for(int i = 0 ; i < 3 ; i++){
            Task_03 task = new Task_03(phaser);
            Thread thread = new Thread(task,"task_" + i);
            thread.start();
        }
        while(!phaser.isTerminated()){
            phaser.arriveAndAwaitAdvance();    //主线程一直等待
        }
        System.out.println("主线程任务已经结束....");
    }

    static class Task_03 implements Runnable{
        private final Phaser phaser;

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

        @Override
        public void run() {
            do{
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + "开始执行任务...");
                phaser.arriveAndAwaitAdvance();
            }while(!phaser.isTerminated());
        }
    }
}

运行结果:

task_0开始执行任务...
task_1开始执行任务...
task_1执行onAdvance方法.....;phase:0registeredParties=3
task_2开始执行任务...
task_0开始执行任务...
task_1开始执行任务...
task_0执行onAdvance方法.....;phase:1registeredParties=3
task_2开始执行任务...
task_2执行onAdvance方法.....;phase:2registeredParties=3
主线程任务已经结束....
task_0开始执行任务...

参考博文:

1、What’s New on Java 7 Phaser

2、http://blog.csdn.net/andycpp/article/details/8838820

时间: 2024-08-06 15:59:05

【Java并发编程实战】—–“J.U.C”:Phaser的相关文章

【Java并发编程实战】—–“J.U.C”:CountDownlatch

上篇博文([Java并发编程实战]-–"J.U.C":CyclicBarrier)LZ介绍了CyclicBarrier.CyclicBarrier所描述的是"允许一组线程互相等待,直到到达某个公共屏障点,才会进行后续任务".而CountDownlatch和它也有一点点相似之处:CountDownlatch所描述的是"在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待".在JDK API中是这样阐述的: 用给定的计数 初始化 Co

【Java并发编程实战】—–“J.U.C”:ReentrantReadWriteLock

ReentrantLock实现了标准的互斥操作,也就是说在某一时刻只有有一个线程持有锁.ReentrantLock采用这种独占的保守锁直接,在一定程度上减低了吞吐量.在这种情况下任何的"读/读"."读/写"."写/写"操作都不能同时发生.然而在实际的场景中我们就会遇到这种情况:有些资源并发的访问中,它大部分时间都是执行读操作,写操作比较少,但是读操作并不影响数据的一致性,如果在进行读操作时采用独占的锁机制,这样势必会大大降低吞吐量.所以如果能够做

【Java并发编程实战】—–“J.U.C”:Semaphore

信号量Semaphore是一个控制访问多个共享资源的计数器,它本质上是一个"共享锁". Java并发提供了两种加锁模式:共享锁和独占锁.前面LZ介绍的ReentrantLock就是独占锁.对于独占锁而言,它每次只能有一个线程持有,而共享锁则不同,它允许多个线程并行持有锁,并发访问共享资源. 独占锁它所采用的是一种悲观的加锁策略,  对于写而言为了避免冲突独占是必须的,但是对于读就没有必要了,因为它不会影响数据的一致性.如果某个只读线程获取独占锁,则其他读线程都只能等待了,这种情况下就限

【Java并发编程实战】—–“J.U.C”:CyclicBarrier

在上篇博客([Java并发编程实战]-–"J.U.C":Semaphore)中,LZ介绍了Semaphore,下面LZ介绍CyclicBarrier.在JDK API中是这么介绍的: 一个同步辅助类,它允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point).在涉及一组固定大小的线程的程序中,这些线程必须不时地互相等待,此时 CyclicBarrier 很有用.因为该 barrier 在释放等待线程后可以重用,所以称它为循环 的 barrier. Cy

【Java并发编程实战】—– AQS(四):CLH同步队列

在[Java并发编程实战]-–"J.U.C":CLH队列锁提过,AQS里面的CLH队列是CLH同步锁的一种变形. 其主要从双方面进行了改造:节点的结构与节点等待机制.在结构上引入了头结点和尾节点,他们分别指向队列的头和尾,尝试获取锁.入队列.释放锁等实现都与头尾节点相关.而且每一个节点都引入前驱节点和后兴许节点的引用:在等待机制上由原来的自旋改成堵塞唤醒. 其结构例如以下: 知道其结构了,我们再看看他的实现.在线程获取锁时会调用AQS的acquire()方法.该方法第一次尝试获取锁假设

《Java并发编程实战》第十章 避免活跃性危险 读书笔记

一.死锁 所谓死锁: 是指两个或两个以上的进程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,它们都将无法推进下去. 百科百科 当两个以上的运算单元,双方都在等待对方停止运行,以取得系统资源,但是没有一方提前退出时,这种状况,就称为死锁.维基百科 1. 顺序死锁 最少有两个锁,一个线程获取到A锁需要获取B锁才能进行操作,而另外一个线程获取到了B锁,需要获取A锁才能执行操作,这种情况下容易出现顺序死锁. public class LeftRightDeadlock { priva

java并发编程实战学习(3)--基础构建模块

转自:java并发编程实战 5.3阻塞队列和生产者-消费者模式 BlockingQueue阻塞队列提供可阻塞的put和take方法,以及支持定时的offer和poll方法.如果队列已经满了,那么put方法将阻塞直到空间可用:如果队列为空,那么take方法将阻塞直到有元素可用.队列可以是有界的也可以是无界的. 如果生产者生成工作的速率比消费者处理工作的速率款,那么工作项会在队列中累计起来,最终好紧内存.同样,put方法的阻塞特性也极大地简化了生产者的编码.如果使用有界队列,当队列充满时,生产者将阻

《java并发编程实战》笔记(一)

最近在看<java并发编程实战>,希望自己有毅力把它读完. 线程本身有很多优势,比如可以发挥多处理器的强大能力.建模更加简单.简化异步事件的处理.使用户界面的相应更加灵敏,但是更多的需要程序猿面对的是安全性问题.看下面例子: public class UnsafeSequence { private int value; /*返回一个唯一的数值*/ public int getNext(){ return value++; } } UnsafeSequence的问题在于,如果执行时机不对,那么

《Java并发编程实战》第十六章 Java内存模型 读书笔记

Java内存模型是保障多线程安全的根基,这里仅仅是认识型的理解总结并未深入研究. 一.什么是内存模型,为什么需要它 Java内存模型(Java Memory Model)并发相关的安全发布,同步策略的规范.一致性等都来自于JMM. 1 平台的内存模型 在架构定义的内存模型中将告诉应用程序可以从内存系统中获得怎样的保证,此外还定义了一些特殊的指令(称为内存栅栏或栅栏),当需要共享数据时,这些指令就能实现额外的存储协调保证. JVM通过在适当的位置上插入内存栅栏来屏蔽在JVM与底层平台内存模型之间的

《Java并发编程实战》第八章 线程池的使用 读书笔记

一.在任务与执行策略之间的隐性解耦 有些类型的任务需要明确地指定执行策略,包括: . 依赖性任务.依赖关系对执行策略造成约束,需要注意活跃性问题.要求线程池足够大,确保任务都能放入. . 使用线程封闭机制的任务.需要串行执行. . 对响应时间敏感的任务. . 使用ThreadLocal的任务. 1. 线程饥饿死锁 线程池中如果所有正在执行任务的线程都由于等待其他仍处于工作队列中的任务而阻塞,这种现象称为线程饥饿死锁. 2. 运行时间较长的任务 Java提供了限时版本与无限时版本.例如Thread