Java并发——Phaser详解

女爷i 2021-08-30 22:00 579阅读 0赞

### Phaser ###

Phaser中文翻译为相位器。它与CountDownLatch非常相似，允许我们协调线程的执行。与CountDownLatch相比，它具有一些额外的功能。

Phaser是在线程动态数需要继续执行之前等待的屏障。在CountDownLatch中，该数字无法动态配置，需要在创建实例时提供。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

arriveAndAwaitAdvanve()方法：  
方`arriveAndAwaitAdvanceo`的作用与 CountDownLatch类中的awai()方法大体一样,通过从方法的名称解释来看, arrive是到达的意思,wait是等待的意思,而 advance是前进、促进的意思,所以执行这个方法的作用就是当前线程已经到达屏障,在此等待一段时间,等条件满足后继续向下一个屏障继续执行。

public class PrintTools { 
        public static Phaser phaser;
    
        public static void methodA() { 
            System.out.println(Thread.currentThread().getName() + " A1 begin=" + System.currentTimeMillis());
            phaser.arriveAndAwaitAdvance();
            System.out.println(Thread.currentThread().getName() + " A1 end=" + System.currentTimeMillis());
    
            System.out.println(Thread.currentThread().getName() + " A2 begin=" + System.currentTimeMillis());
            phaser.arriveAndAwaitAdvance();
            System.out.println(Thread.currentThread().getName() + " A2 end=" + System.currentTimeMillis());
        }
    
        public static void methodB() { 
            try { 
                System.out.println(Thread.currentThread().getName() + " A1 begin=" + System.currentTimeMillis());
                phaser.arriveAndAwaitAdvance();
                System.out.println(Thread.currentThread().getName() + " A1 end=" + System.currentTimeMillis());
    
                System.out.println(Thread.currentThread().getName() + " A2 begin=" + System.currentTimeMillis());
                Thread.sleep(5000);
                phaser.arriveAndAwaitAdvance();
                System.out.println(Thread.currentThread().getName() + " A2 end=" + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) { 
            Phaser p = new Phaser(3);
            PrintTools.phaser = p;
    
            ThreadA a = new ThreadA(phaser);
            a.setName("A");
            a.start();
    
            ThreadB b = new ThreadB(phaser);
            b.setName("B");
            b.start();
    
            ThreadC c = new ThreadC(phaser);
            c.setName("C");
            c.start();
        }
    }

public class ThreadA extends Thread { 
        private Phaser phaser;
    
        public ThreadA(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            PrintTools.methodA();
        }
    }
    public class ThreadB extends Thread { 
        private Phaser phaser;
    
        public ThreadB(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            PrintTools.methodA();
        }
    }
    public class ThreadC extends Thread { 
        private Phaser phaser;
    
        public ThreadC(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            PrintTools.methodB();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]  
`注意：当arriveAndAwaitAdvance()调用时，如果计数不足，则线层呈阻塞状态，不继续向下运行`  
修改methodB()方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]  
再次运行，输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]

arriveAndDeregister()方法：  
当线程到达屏障后，退出屏障的阻塞，并且使parties值减一。

public class Deregister { 
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(2);
            MyThread t = new MyThread(phaser);
            t.start();
            System.out.println("main begin" + System.currentTimeMillis());
            System.out.println(phaser.getRegisteredParties());
            phaser.arriveAndDeregister();//退出阻塞，计数-1
            System.out.println(phaser.getRegisteredParties());
            System.out.println("main end" + System.currentTimeMillis());
        }
    }
    
    class MyThread extends Thread{ 
        private Phaser phaser;
    
        public MyThread(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            phaser.arriveAndAwaitAdvance();
            System.out.println("MyThread end");
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

getPhase()和onAdvance()：

方法getPhase()获取的是已经到达第几个屏障。

public class getPhase { 
    
        public static void main(String[] args) throws InterruptedException { 
            Phaser phaser = new Phaser(1);
            TestThread t = new TestThread(phaser);
            t.start();
        }
    
    
    }
    class TestThread extends Thread { 
        private Phaser phaser;
    
        public TestThread(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            System.out.println("A begin");
            phaser.arriveAndAwaitAdvance();
            System.out.println("A end phase value=" + phaser.getPhase());
    
            System.out.println("A begin");
            phaser.arriveAndAwaitAdvance();
            System.out.println("A end phase value=" + phaser.getPhase());
    
            System.out.println("A begin");
            phaser.arriveAndAwaitAdvance();
            System.out.println("A end phase value=" + phaser.getPhase());
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

`onAdvance()的作用是通过新的屏障时被调用：`

public class onAdvance { 
        public static void main(String[] args) throws InterruptedException { 
            Phaser phaser = new Phaser(2) { 
                @Override
                protected boolean onAdvance(int phase, int registeredParties) { 
                    System.out.println("onAdvence被调用！");
                    return true;
                    //返回true表示，下一次到达屏障不等待了，Phaser呈无效/销毁的状态
                    //返回false表示，下一次到达屏障Phaser继续工作
                }
            };
    
            //第一次通过屏障
            AThread a = new AThread(phaser);
            a.start();
            phaser.arriveAndAwaitAdvance();
    
    
            //第二次
            Thread.sleep(2000);
            BThread b = new BThread(phaser);
            b.start();
            System.out.println("main end");
        }
    }
    
    class AThread extends Thread { 
        private Phaser phaser;
    
        public AThread(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            System.out.println("A begin");
            phaser.arriveAndAwaitAdvance();
            System.out.println("A end");
        }
    }
    
    class BThread extends Thread { 
        private Phaser phaser;
    
        public BThread(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            System.out.println("B begin");
            phaser.arriveAndAwaitAdvance();
            System.out.println("B end");
        }
    }

onAdvance返回true，下一次到达屏障不阻塞  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

修改返回值为false，下一次到达屏障正常阻塞  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]

getRegisteredParities():  
获得注册的parties数量。

bulkRegister()：  
批量增加parties数量。

getArrivedParties():  
获得已经被使用的parties个数。

getUnarrivedParties()：  
获得未被使用的parties个数。

public class Parties { 
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(5);
            System.out.println(phaser.getRegisteredParties());//5
    
            phaser.register();
            System.out.println(phaser.getRegisteredParties());//6
    
            phaser.bulkRegister(10);
            System.out.println(phaser.getRegisteredParties());//16
    
            System.out.println(phaser.getArrivedParties());//0
            System.out.println(phaser.getUnarrivedParties());//16
        }
    }

arrive()方法：  
方法arrive()的作用是使parties 值加1，并且不在屏障处等待，直接向下面的代码继续运行，并且Phaser类有计数重置功能。

public class TestArrive { 
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(2){ 
                @Override
                protected boolean onAdvance(int phase, int registeredParties) { 
                    System.out.println("到达屏障，但未通过！phase=" + phase + " registeredParties=" + registeredParties);
                    return super.onAdvance(phase, registeredParties);
                }
            };
    
            System.out.println("A1 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
            phaser.arrive();
            System.out.println("A1 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
    
            System.out.println("A2 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
            phaser.arrive();
            System.out.println("A2 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
    
            System.out.println("B1 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
            phaser.arrive();
            System.out.println("B1 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
    
            System.out.println("B2 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
            phaser.arrive();
            System.out.println("B2 getPhase=" + phaser.getPhase() + " getRegisteredParties=" + phaser.getRegisteredParties() +
                    " getArrivedParties=" + phaser.getArrivedParties());
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]  
方法arrive()的功能是使getArrivedParties()计数加1，不等待其他线程到达屏障。  
在控制台中多次出现getArrivedParties=0的运行结果，所以可以分析出Phaser类在经过屏障点后计数能被重置。

awaitAdvance(int phase):  
如果传人参数phase值和当前getPhase()方法返回值一样，则在屏障处等待，否则继续向下面运行，有些类似于旁观者的作用，当观察的条件满足了就等待(旁观),如果条件不满足，则程序向下继续运行。

public class AwaitAdvance { 
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(1);
            phaser.awaitAdvance(1);//不一样，通过
            System.out.println("通过第一次");
    
            phaser.arriveAndAwaitAdvance();
            phaser.awaitAdvance(1);//一样，阻塞
            System.out.println("通过第二次");
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 9]

awaitAdvanceInterruptibly(int):  
可中断的。  
awaitAdvanceInterruptibly(int ,long, TImeUnit):  
带超时，且可中断。

forceTermination()和isTerminated():  
方法`forceTermination()`使Phaser对象的屏障功能失效，而方法`isTerminated()`是判断Phaser对象是否已经呈销毁状态。

public class Termination { 
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(2);
            phaser.forceTermination();
            phaser.arriveAndAwaitAdvance();//失效，所以不阻塞
            System.out.println(phaser.isTerminated());//true
        }
    }

#### 控制Phaser类的运行时机 ####

前面的示例都是线程一起到达屏障后继续运行，有些情况下是需要进行控制的，也就是到达屏障后不允许继续运行。  
默认的运行效果：

public class RunDemo { 
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(3);
            for (int i = 0; i < 3; i++) { 
                RunThread t = new RunThread(phaser);
                t.start();
            }
        }
    }
    
    class RunThread extends Thread { 
        private Phaser phaser;
    
        public RunThread(Phaser phaser) { 
            this.phaser = phaser;
        }
    
        @Override
        public void run() { 
            System.out.println(Thread.currentThread().getName() + " A1 begin=" + System.currentTimeMillis());
            phaser.arriveAndAwaitAdvance();
            System.out.println(Thread.currentThread().getName() + " A1 end=" + System.currentTimeMillis());
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 10]

可控的运行时机：

public static void main(String[] args) throws InterruptedException { 
            Phaser phaser = new Phaser(3);
            phaser.register();
            for (int i = 0; i < 3; i++) { 
                RunThread t = new RunThread(phaser);
                t.start();
            }
            Thread.sleep(2000);
            phaser.arriveAndAwaitAdvance();
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 11]  
此实验说明运行的时机是可以通过逻辑控制的，主要的原理就是计数+1,然后通过逻辑代码的方式来决定线程是否继续向下运行。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/fcf10dfd38624a068712880d4c91528b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/12891f7471004263a20a8dff1f662ad5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/45a7fd8e9b724179839a2f398f48e555.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/dcb4044bab104173942d07e31d90bd52.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/8e2f9dc8c0534834a62ac308da9e1d77.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/2a5a7189b7d549968c836ffb0604d593.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/938814202c454c488187ca15e836f967.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]: /images/20210813/2fc26a7c3ed54a4b8b9bb68e8e857610.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]: /images/20210813/78e75b3ccdfd4bb98bbadc242aceb25c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 9]: /images/20210813/f0bc49dd9ae048229c3d3acd3d1184b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 10]: /images/20210813/d72b9a9e80f647eba3a791825907c41c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 11]: /images/20210813/af4d797fe22f467e849c00cd232a0edb.png