三.多线程JUC篇-3.13 Phaser工具

比眉伴天荒 2022-12-21 00:59 175阅读 0赞

## 1.前言 ##

Phaser是JDK1.7提供的一种同步工具，是可重用的同步屏障，其功能类似于CyclicBarrier和CountDownLatch，但支持更灵活的用法：可以分阶段控制各个线程的行为，可以动态的创建也可以动态的注销。

## 2.用法 ##

### 2.1 动态注册 ###

public class MyPhaser7 { 
    
        public static final Random random = new Random(System.currentTimeMillis());
    
        public static void main(String[] args) { 
            final Phaser phaser = new Phaser();
    
            IntStream.rangeClosed(1, 5).boxed().map(i -> phaser).forEach(Task::new);
    
            phaser.register();
            phaser.arriveAndAwaitAdvance(); 
            System.out.println("All of workers finished the task.");
        }
    
        private static class Task extends Thread{ 
    
            private final Phaser phaser;
    
            public Task(Phaser phaser) { 
                this.phaser = phaser;
                this.phaser.register();
                start();
            }
    
            @Override
            public void run() { 
                System.out.println("The worker [" + getName() + "] is working..." );
    
                try { 
                    TimeUnit.SECONDS.sleep(random.nextInt(10));
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
    
                this.phaser.arriveAndAwaitAdvance();
            }
        }
    }
    
    ----------------------------
    The worker [Thread-0] is working...
    The worker [Thread-2] is working...
    The worker [Thread-4] is working...
    The worker [Thread-1] is working...
    The worker [Thread-3] is working...
    All of workers finished the task.

### 2.2 多轮次 ###

/** * 每个run()里面的phaser.arriveAndAwaitAdvance()都是一轮getPhase()的消费 * phaser的同一阶段arriveAndAwaitAdvance()全部执行完成才会执行下一阶段，即是相同的getPhase()值时 */
    public class MyPhaser8 { 
        private static final Random random = new Random(System.currentTimeMillis());
    
        static class Athletes extends Thread { 
            private final int num;
            private final Phaser phaser;
    
            public Athletes(int num, Phaser phaser) { 
                this.num = num;
                this.phaser = phaser;
            }
    
            @Override
            public void run() { 
                try { 
                    System.out.println(num + ": start running...");
                    TimeUnit.SECONDS.sleep(random.nextInt(5));
                    System.out.println(num + ": end running...");
                    System.out.println(num + ": getPhase()=>" + phaser.getPhase());
                    phaser.arriveAndAwaitAdvance();
    
                    System.out.println(num + ": start bicycle...");
                    TimeUnit.SECONDS.sleep(random.nextInt(5));
                    System.out.println(num + ": end bicycle...");
                    System.out.println(num + ": getPhase()=>" + phaser.getPhase());
                    phaser.arriveAndAwaitAdvance();
    
                    System.out.println(num + ": start long jump...");
                    TimeUnit.SECONDS.sleep(random.nextInt(5));
                    System.out.println(num + ": end long jump...");
                    System.out.println(num + ": getPhase()=>" + phaser.getPhase());
                    phaser.arriveAndAwaitAdvance();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
        public static void main(String[] args) { 
            final Phaser phaser = new Phaser(5);
    
            for(int i = 0; i < 5; i++) { 
                new Athletes(i, phaser).start();
            }
        }
    }

### 2.3 可以动态的创建也可以动态的注销 ###

/** * Phaser可以动态的创建也可以动态的注销 * 相当于一轮一轮的动作完成 * * 由于一个运动员受伤了， long jump这个动作没有完成，其它运动员都在等待他完成 * 导致线程不能终止。 因此这个运动员需要phaser.arriveAndDeregister()，取消该相位注册让其他的相位 * 不再等待 */
    public class MyPhaser9 { 
        private static final Random random = new Random(System.currentTimeMillis());
    
        static class InjuredAthletes extends Thread { 
            private final int num;
            private final Phaser phaser;
    
            public InjuredAthletes(int num, Phaser phaser) { 
                this.num = num;
                this.phaser = phaser;
            }
    
            @Override
            public void run() { 
                try { 
                    sport(phaser, num + ": start running...", num + ": end running...");
    
                    sport(phaser, num + ": start bicycle...", num + ": end bicycle...");
    
    // System.out.println("Oh, shit. I am injured.");
    
                    System.out.println("Oh shit, i am injured, i will exited.");
                    phaser.arriveAndDeregister();//取消该相位注册让其他的相位不再等待
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        static class Athletes extends Thread { 
            private final int num;
            private final Phaser phaser;
    
            public Athletes(int num, Phaser phaser) { 
                this.num = num;
                this.phaser = phaser;
            }
    
            @Override
            public void run() { 
                try { 
                    sport(phaser, num + ": start running...", num + ": end running...");
    
                    sport(phaser, num + ": start bicycle...", num + ": end bicycle...");
    
                    sport(phaser, num + ": start long jump...", num + ": end long jump...");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        private static void sport(Phaser phaser, String str1, String str2) throws InterruptedException { 
            System.out.println(str1);
            TimeUnit.SECONDS.sleep(random.nextInt(5));
            System.out.println(str2);
            phaser.arriveAndAwaitAdvance();
        }
    
        public static void main(String[] args) { 
            final Phaser phaser = new Phaser(5);
    
            for(int i = 1; i < 5; i++) { 
                new Athletes(i, phaser).start();
            }
    
            new InjuredAthletes(5, phaser).start();
        }
    }

### 2.4 arriveAndAwaitAdvance ###

报告已到达，且要求所有的parties到达才可以继续运行，否则阻塞等待

public class MyPhaser { 
        public static void main(String[] args) { 
            final Phaser phaser = new Phaser(3); //声明了三个parities
            new Thread(phaser::arriveAndAwaitAdvance).start(); 
    
            try { 
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
    
            System.out.println("registered parties: " + phaser.getRegisteredParties());
    
            System.out.println("arrived parties: " + phaser.getArrivedParties());
            System.out.println("unarrived parties: " + phaser.getUnarrivedParties());
    		
    		//现在只有一个parties到达，需要等待，所以主线程会永远等待下去
        }
    }
    
    ---------------------
    registered parties: 3
    arrived parties: 1
    unarrived parties: 2

### 2.5 arrive ###

报告已到达，立即返回，不阻塞等待（不需要检查所有的parties到达）  
有一个场景：子任务分二个阶段，主线程只需要了解第一阶段的所有任务完成，不需要监控任务的第二阶段

public class MyPhaser3 { 
        private final static Random random = new Random(System.currentTimeMillis());
        public static void main(String[] args) { 
            Phaser phaser = new Phaser(5);
            for (int i = 0; i< 4; i++){ 
                new ArriveTask(phaser, i+1).start();
            }
    
            phaser.arriveAndAwaitAdvance();
            System.out.println("all task phase 1 end.");
    
    
        }
    
        private static class ArriveTask extends Thread{ 
            private final Phaser phaser;
    
            public ArriveTask(Phaser phaser, int no) { 
                super(String.valueOf(no));
                this.phaser = phaser;
            }
    
            @Override
            public void run() { 
                System.out.println(getName() + " start work...");
    
                mysleep();
                this.phaser.arrive();
                System.out.println(getName() + " phase 1 end.");
    
                mysleep();
                System.out.println(getName() + " phase 2 end.");
            }
        }
    
        private static void mysleep(){ 
            try { 
                TimeUnit.SECONDS.sleep(random.nextInt(5));
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }
    
    ------------------------------------
    1 start work...
    3 start work...
    3 phase 1 end.
    2 start work...
    4 start work...
    1 phase 1 end.
    4 phase 1 end.
    3 phase 2 end.
    2 phase 1 end.
    all task phase 1 end.
    2 phase 2 end.
    1 phase 2 end.
    4 phase 2 end.

### 2.6 awaitAdvance ###

(1) 不会报告已到达

public class MyPhaser4 { 
        public static void main(String[] args) { 
            final Phaser phaser = new Phaser(3);
            new Thread(()->phaser.awaitAdvance(phaser.getPhase())).start();
    
            try { 
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            System.out.println("arrived parties: " +phaser.getArrivedParties());
        }
    }
    ----------------
    arrived parties: 0

(2) 传入的Phase，与当前实际的Phase一致，则阻塞住；与当前实际的Phase不一致，立即返回

public static void main(String[] args) { 
            final Phaser phaser = new Phaser(3);
            // phaser.awaitAdvance(phaser.getPhase()); //阻塞，不会打印end
            //phaser.awaitAdvance(2); //不阻塞，会打印end
            System.out.println("end");
        }

(3) 虽然主线程不参于报到，但依然可以阻塞等待所有任务

public class MyPhaser4 { 
        private final static Random random = new Random(System.currentTimeMillis());
    
        public static void main(String[] args) { 
            final Phaser phaser = new Phaser(3);
            IntStream.rangeClosed(0, 3).boxed().map(i -> phaser).forEach(ArriveTask::new);
    
            phaser.awaitAdvance(phaser.getPhase()); 
            System.out.println("=====");
        }
    
        private static class ArriveTask extends Thread { 
            private final Phaser phaser;
    
            public ArriveTask(Phaser phaser) { 
                this.phaser = phaser;
                start();
            }
    
            @Override
            public void run() { 
                System.out.println(getName() + " start...");
                mysleep();
                phaser.arriveAndAwaitAdvance();
                System.out.println(getName() + " end");
            }
        }
    
        private static void mysleep(){ 
            try { 
                TimeUnit.SECONDS.sleep(random.nextInt(5));
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }
    ------
    Thread-0 start...
    Thread-2 start...
    Thread-3 start...
    Thread-1 start...
    Thread-2 end
    =====
    Thread-1 end
    Thread-0 end

### 2.7 awaitAdvanceInterruptibly ###

可中断的awaitAdvance

public class MyPhaser5 { 
        public static void main(String[] args) { 
            final Phaser phaser = new Phaser(3);
    
            Thread thread =  new Thread(()->{ 
                try { 
                    phaser.awaitAdvanceInterruptibly(phaser.getPhase());
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
    
            thread.start();
    
            try { 
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
    
            thread.interrupt();
    
        }
    }

### 2.8 forceTermination ###

强制结束phaser

public class MyPhaser6 { 
        public static void main(String[] args) throws InterruptedException { 
            final Phaser phaser = new Phaser(3);
    
            new Thread(phaser::arriveAndAwaitAdvance).start();
    
            TimeUnit.SECONDS.sleep(3);
            System.out.println("phaser state: " +  phaser.isTerminated());
    
            phaser.forceTermination();
            System.out.println("phaser state: " +  phaser.isTerminated());
    
        }
    }
    
    ----------------------
    phaser state: false
    phaser state: true

### 2.9 综合例子 ###

总共六位同学要参加同学聚会，聚会流程有三项：吃饭、K歌、看电影。六位同学都要吃饭，但是吃完饭后只有四位同学想K歌，K完歌后只有两位同学想看电影。实现这样的场景，用CyclicBarrier和CountDownLatch显然不太方便。这时就可以使用Phaser。

Phaser定义流程：  
     首先定义聚会流程，定义子类PartyPhaser来实现Phaser，并且重写onAdvance方法

import java.util.concurrent.Phaser;
    
    /** * @author sicimike */
    public class PartyPhaser extends Phaser { 
    
        @Override
        protected boolean onAdvance(int phase, int registeredParties) { 
            switch (phase) { 
                case 0:
                	// 所有注册的线程都到达第一个屏障后，执行这个case
                    System.out.println("所有人均已吃完，共：" + registeredParties + " 人");
                    return false;
                case 1:
                	// 所有注册的线程都到达第二个屏障后，执行这个case
                    System.out.println("所有人均已达到KTV，共：" + registeredParties + " 人");
                    return false;
                case 2:
                	// 所有注册的线程都到达第三个屏障后，执行这个case
                    System.out.println("所有人均已到达电影院，共：" + registeredParties + " 人");
                    return true;
                default:
                    return true;
            }
        }
    }