JUC工具类Phaser

「爱情、让人受尽委屈。」 2022-08-28 13:51 195阅读 0赞

# 简介 #

java7中引入了一种新的可重复使用的同步屏障,Phaser阶段器，与CyclicBarrier和CountDownLatch类似，但更强大。

CyclicBarrier解决了CountDownLatch不能重用的问题，但是仍有以下不足：

1）不能动态调整计数器值，假如线程数不足以打破barrier，就只能reset或者多加些线程，在实际运用中显然不现实

2）每次await仅消耗1个计数器值，不够灵活

Phaser就是用来解决这些问题的。Phaser将多个线程协作执行的任务划分为多个阶段，每个阶段都可以有任意个参与者，线程可以随时注册并参与到某个阶段。

# Phaser的常用方法 #

**1、register方法**  
动态添加一个parties

`int register()`

**2、bulkRegister方法**  
动态添加多个parties

parties：需要添加的个数  
`int bulkRegister(int parties)`

**3、getRegisteredParties方法**  
获取当前的parties数

`int getRegisteredParties()`

**4、arriveAndAwaitAdvance方法**  
到达并等待其他线程到达

`int arriveAndAwaitAdvance()`

**5、arriveAndDeregister方法**  
到达并注销该parties，这个方法不会使线程阻塞

`int arriveAndDeregister()`

**6、arrive方法**  
到达，但不会使线程阻塞

`int arrive()`

**7、awaitAdvance方法**  
等待前行，可阻塞也可不阻塞，判断条件为传入的phase是否为当前phaser的phase。如果相等则阻塞，反之不进行阻塞

phase：阶段数值  
`int awaitAdvance(int phase)`

**8、awaitAdvanceInterruptibly方法**  
该方法与awaitAdvance类似，唯一不一样的就是它可以进行打断。

phase：阶段数值  
timeout：超时时间  
unit：时间单位  
`int awaitAdvanceInterruptibly(int phase)`  
`int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)`

9、getArrivedParties方法  
获取当前到达的parties数

`int getArrivedParties()`

**10、getUnarrivedParties方法**  
获取当前未到达的parties数

`int getUnarrivedParties()`

**11、getPhase方法**  
获取当前属于第几阶段，默认从0开始，最大为integer的最大值

`int getPhase()`

**12、isTerminated方法**  
判断当前phaser是否关闭

`boolean isTerminated()`

**13、forceTermination方法**  
强制关闭当前phaser

`void forceTermination()`

## **常用方法演示** ##

package com.dongguo.juc;
    
    import java.util.concurrent.Phaser;
    import java.util.concurrent.TimeUnit;
    
    /** * @author Dongguo * @date 2021/9/19 0019-13:26 * @description: */
    public class PhaserTest2 { 
        public static void main(String[] args) throws InterruptedException { 
            //初始化5个parties
            Phaser phaser = new Phaser(5);
    
            //只有当全部线程通过时才会进入下一阶段，从0开始
            System.out.println("当前阶段数："+phaser.getPhase());
    
            //添加一个parties
            phaser.register();
            System.out.println("当前Parties数："+phaser.getRegisteredParties());
            //添加多个parties
            phaser.bulkRegister(4);
            System.out.println("当前Parties数："+phaser.getRegisteredParties());
    
            new Thread(()->{ 
                //到达并等待其他线程到达
                phaser.arriveAndAwaitAdvance();
            },"t1").start();
            
            new Thread(()->{ 
                //到达后注销该parties，不等待其他线程
                phaser.arriveAndDeregister();
                System.out.println("go on");
            },"t2").start();
    
            TimeUnit.MILLISECONDS.sleep(100);
            System.out.println("当前Parties数："+phaser.getRegisteredParties());
            System.out.println("当前到达数："+phaser.getArrivedParties());
            System.out.println("当前未达数："+phaser.getUnarrivedParties());
    
            //何时会停止，只有当parties中的数量为0时或者调用forceTermination方法就会停止了，我们也可以重写phaser中的onAdvance，给他返回true就会使这个phaser停止了
            System.out.println("phaser是否结束："+phaser.isTerminated());
            phaser.forceTermination();
            System.out.println("phaser是否结束："+phaser.isTerminated());
        }
    }
    运行结果
    当前阶段数：0
    当前Parties数：6
    当前Parties数：10
    go on
    当前Parties数：9
    当前到达数：1
    当前未达数：8
    phaser是否结束：false
    phaser是否结束：true

# 案例 #

## 使用Phaser动态注册parties ##

package com.dongguo.juc;
    
    import java.util.concurrent.Phaser;
    import java.util.concurrent.TimeUnit;
    
    /** * @author Dongguo * @date 2021/9/19 0019-11:40 * @description:5个工人全都完成工作，同一时刻完成 */
    public class PhaserTest { 
    
        private static Phaser phaser = new Phaser();
    
        public static void main(String args[]) { 
            for (int i = 1; i <= 5; i++) { 
              new Thread(()->{ 
                  phaser.register();
    
                  System.out.println(Thread.currentThread().getName() +" is working");
                  try { 
                      TimeUnit.SECONDS.sleep(2);
                  } catch (InterruptedException e) { 
                      e.printStackTrace();
                  }
                  phaser.arriveAndAwaitAdvance();//等待
                  System.out.println(Thread.currentThread().getName() +" work finished "+System.currentTimeMillis());
              },"t"+i).start();
            }
        }
    }
    运行结果
    t1 is working
    t3 is working
    t2 is working
    t4 is working
    t5 is working
    t4 work finished 1632029871265
    t5 work finished 1632029871265
    t2 work finished 1632029871265
    t1 work finished 1632029871265
    t3 work finished 1632029871265

## 使用Phaser设置多个阶段 ##

package com.dongguo.juc;
    
    import java.util.Random;
    import java.util.concurrent.Phaser;
    import java.util.concurrent.TimeUnit;
    
    /** * @author Dongguo * @date 2021/9/19 0019-12:42 * @description: 模拟3个运动员一同参加3项运动 */
    public class PhaserTest1 { 
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(3);
    
            for (int i = 1; i <= 3; i++) { 
                int no =i;
                new Thread(() -> { 
                    try { 
                        System.out.println(no+": 当前处于第："+phaser.getPhase()+"阶段");
                        System.out.println(no+": start running");
                        TimeUnit.SECONDS.sleep(2);
                        System.out.println(no+": end running");
                        //等待其他运动员完成跑步
                        phaser.arriveAndAwaitAdvance();
    
                        System.out.println(no+": 当前处于第："+phaser.getPhase()+"阶段");
                        System.out.println(no+": start bicycle");
                        TimeUnit.SECONDS.sleep(2);
                        System.out.println(no+": end bicycle");
                        //等待其他运动员完成骑行
                        phaser.arriveAndAwaitAdvance();
    
                        System.out.println(no+": 当前处于第："+phaser.getPhase()+"阶段");
                        System.out.println(no+": start long jump");
                        TimeUnit.SECONDS.sleep(2);
                        System.out.println(no+": end long jump");
                        //等待其他运动员完成跳远
                        phaser.arriveAndAwaitAdvance();
    
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }, "t"+i).start();
            }
        }
    }
    运行结果
    1: 当前处于第：0阶段
    2: 当前处于第：0阶段
    2: start running
    1: start running
    3: 当前处于第：0阶段
    3: start running
    2: end running
    1: end running
    3: end running
    3: 当前处于第：1阶段
    3: start bicycle
    1: 当前处于第：1阶段
    2: 当前处于第：1阶段
    1: start bicycle
    2: start bicycle
    1: end bicycle
    2: end bicycle
    3: end bicycle
    1: 当前处于第：2阶段
    1: start long jump
    3: 当前处于第：2阶段
    3: start long jump
    2: 当前处于第：2阶段
    2: start long jump
    1: end long jump
    3: end long jump
    2: end long jump

# 总结 #

Phaser 可以通过register() 方法和arriveAndDeregister() 方法，动态的增加或者减少注册量  
使用arriveAndAwaitAdvance，相当于CyclicBarrier机制  
使用arrive，相当于countdown机制  
可以利用awaitAdvance，让主线程等待子线程全部完成任务