public class PhaserBatchProcessorDemo {

    private final List<String> data;
    private final int batchsize;//一次處理多少數(shù)據(jù)
    private final int threadCount;//處理的線程數(shù)
    private final Phaser phaser;
    private final List<String> processedData;

    public PhaserBatchProcessorDemo(List<String> data,int batchsize,int threadCount){
        this.data = data;
        this.batchsize = batchsize;
        this.threadCount = threadCount;
        this.phaser = new Phaser(1);
        //this.phaser = new Phaser();
        this.processedData = new ArrayList<>();
    }

    public void process() throws InterruptedException {
        for(int i = 0;i<threadCount;i++){
            System.out.println("phaser.register():"+phaser.register());
            new Thread(new BatchProcessor(i)).start();
            Thread.sleep(50);
        }
        phaser.arriveAndDeregister();//主線程執(zhí)行結(jié)束
        System.out.println("結(jié)束");
    }

    private class BatchProcessor implements Runnable{
        private final int threadIndex;
        public BatchProcessor(int threadIndex){this.threadIndex = threadIndex;}

        @Override
        public void run() {
            int index = 0;
            while(true){
                //所有線程都到達(dá)這個(gè)點(diǎn)之前會(huì)阻塞
                System.out.println("線程"+threadIndex+"phaser.arriveAndAwaitAdvance1():");
                phaser.arriveAndAwaitAdvance();

                //從未處理數(shù)據(jù)中找到一個(gè)可以處理的批次
                List<String> batch = new ArrayList<>();
                synchronized (data){
                    while (index < data.size()&&batch.size()<batchsize){
                        String d = data.get(index);
                        if(!processedData.contains(d)){
                            batch.add(d);
                            processedData.add(d);
                        }
                        index++;
                    }
                }
                //處理數(shù)據(jù)
                for(String d:batch){
                    System.out.println("線程"+threadIndex+"處理數(shù)據(jù)"+d);
                }
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                //所有數(shù)據(jù)都處理完當(dāng)前批次之前會(huì)阻塞
                System.out.println("線程"+threadIndex+"phaser.arriveAndAwaitAdvance2():");
                phaser.arriveAndAwaitAdvance();
                //所有線程都處理完當(dāng)前批次并且未處理數(shù)據(jù)已經(jīng)處理完之前會(huì)阻塞
                if(batch.isEmpty()||index >= data.size()){
                    System.out.println("線程"+threadIndex+"phaser.arriveAndDeregister()"+phaser.arriveAndDeregister());
                    break;
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        List<String> data = new ArrayList<>();
        for(int i = 1;i<=15;i++){
            data.add(String.valueOf(i));
        }

        PhaserBatchProcessorDemo processor = new PhaserBatchProcessorDemo(data,4,3);
        processor.process();
    }
}

/**
 * phaser.register():0
 * 線程0phaser.arriveAndAwaitAdvance1():
 * phaser.register():0
 * 線程1phaser.arriveAndAwaitAdvance1():
 * phaser.register():0
 * 線程2phaser.arriveAndAwaitAdvance1():
 * 結(jié)束
 * 線程2處理數(shù)據(jù)9
 * 線程2處理數(shù)據(jù)10
 * 線程1處理數(shù)據(jù)5
 * 線程1處理數(shù)據(jù)6
 * 線程1處理數(shù)據(jù)7
 * 線程1處理數(shù)據(jù)8
 * 線程2處理數(shù)據(jù)11
 * 線程2處理數(shù)據(jù)12
 * 線程0處理數(shù)據(jù)1
 * 線程0處理數(shù)據(jù)2
 * 線程0處理數(shù)據(jù)3
 * 線程0處理數(shù)據(jù)4
 * 線程2phaser.arriveAndAwaitAdvance2():
 * 線程0phaser.arriveAndAwaitAdvance2():
 * 線程1phaser.arriveAndAwaitAdvance2():
 * 線程2phaser.arriveAndAwaitAdvance1():
 * 線程1phaser.arriveAndAwaitAdvance1():
 * 線程0phaser.arriveAndAwaitAdvance1():
 * 線程0處理數(shù)據(jù)13
 * 線程0處理數(shù)據(jù)14
 * 線程0處理數(shù)據(jù)15
 * 線程0phaser.arriveAndAwaitAdvance2():
 * 線程1phaser.arriveAndAwaitAdvance2():
 * 線程2phaser.arriveAndAwaitAdvance2():
 * 線程1phaser.arriveAndDeregister()4
 * 線程2phaser.arriveAndDeregister()4
 * 線程0phaser.arriveAndDeregister()4
 */

public class PhaserDemo {
    public static void main(String[] args) {
        final Phaser phaser = new Phaser(){
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                //參與者數(shù)量，去除主線程
                int persons = registeredParties - 1;
                switch (phase){
                    case 0:
                        System.out.println("大家都到佘山站了，可以出發(fā)去佘山了，人數(shù)："+persons);
                        break;
                    case 1:
                        System.out.println("大家都到佘山了，出發(fā)去爬山，人數(shù)："+persons);
                        break;
                    case 2:
                        System.out.println("大家都到山頂了，開(kāi)始休息，人數(shù)："+persons);
                        break;

                }
                //判斷是否只剩下一個(gè)主線程，如果是，返回true,代表終止
                return registeredParties ==1;
            }
        };

        phaser.register();
        final PersonTask personTask = new PersonTask();
        //3個(gè)全程參加的伙伴
        for(int i = 0;i<3;i++){

            phaser.register();
            new Thread(()->{
                try{
                    personTask.step1Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step2Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step3Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step4Task();
                    phaser.arriveAndDeregister();


                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }).start();
        }
        

        //兩個(gè)在山腰半路返回
        for(int i = 0;i<2;i++){
            phaser.register();
            new Thread(()->{
                try{
                    personTask.step1Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step2Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step3Task();
                    System.out.println("伙伴【"+Thread.currentThread().getName()+"】中途山腰返回");
                    phaser.arriveAndDeregister();

                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }).start();
        }

        while (!phaser.isTerminated()) {
            int phase = phaser.arriveAndAwaitAdvance();
            if (phase == 2) {
                //兩個(gè)在佘山直接會(huì)合
                for(int i = 0;i<2;i++){
                    phaser.register();
                    new Thread(()->{
                        try{
                            System.out.println("伙伴【"+Thread.currentThread().getName()+"】中途加入");

                            personTask.step3Task();
                            phaser.arriveAndAwaitAdvance();

                            personTask.step4Task();
                            phaser.arriveAndDeregister();

                        }catch (InterruptedException e){
                            e.printStackTrace();
                        }
                    }).start();
                }
            }
        }
    }

    static final Random random = new Random();

    static class PersonTask{
        public void step1Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"從家出發(fā)了......");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"到達(dá)佘山站");
        }

        public void step2Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"出發(fā)去佘山");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"到達(dá)佘山");
        }

        public void step3Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"出發(fā)去爬山");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"到達(dá)山頂");
        }

        public void step4Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"開(kāi)始休息");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"休息結(jié)束，下山回家");
        }
    }
}

/**
 * 伙伴【Thread-3】從家出發(fā)了......
 * 伙伴【Thread-1】從家出發(fā)了......
 * 伙伴【Thread-2】從家出發(fā)了......
 * 伙伴【Thread-4】從家出發(fā)了......
 * 伙伴【Thread-0】從家出發(fā)了......
 * 伙伴【Thread-1】到達(dá)佘山站
 * 伙伴【Thread-4】到達(dá)佘山站
 * 伙伴【Thread-3】到達(dá)佘山站
 * 伙伴【Thread-0】到達(dá)佘山站
 * 伙伴【Thread-2】到達(dá)佘山站
 * 大家都到佘山站了，可以出發(fā)去佘山了，人數(shù)：5
 * 伙伴【Thread-3】出發(fā)去佘山
 * 伙伴【Thread-2】出發(fā)去佘山
 * 伙伴【Thread-4】出發(fā)去佘山
 * 伙伴【Thread-1】出發(fā)去佘山
 * 伙伴【Thread-0】出發(fā)去佘山
 * 伙伴【Thread-1】到達(dá)佘山
 * 伙伴【Thread-4】到達(dá)佘山
 * 伙伴【Thread-3】到達(dá)佘山
 * 伙伴【Thread-0】到達(dá)佘山
 * 伙伴【Thread-2】到達(dá)佘山
 * 大家都到佘山了，出發(fā)去爬山，人數(shù)：5
 * 伙伴【Thread-1】出發(fā)去爬山
 * 伙伴【Thread-4】出發(fā)去爬山
 * 伙伴【Thread-0】出發(fā)去爬山
 * 伙伴【Thread-2】出發(fā)去爬山
 * 伙伴【Thread-3】出發(fā)去爬山
 * 伙伴【Thread-6】中途加入
 * 伙伴【Thread-6】出發(fā)去爬山
 * 伙伴【Thread-5】中途加入
 * 伙伴【Thread-5】出發(fā)去爬山
 * 伙伴【Thread-2】到達(dá)山頂
 * 伙伴【Thread-5】到達(dá)山頂
 * 伙伴【Thread-0】到達(dá)山頂
 * 伙伴【Thread-4】到達(dá)山頂
 * 伙伴【Thread-4】中途山腰返回
 * 伙伴【Thread-6】到達(dá)山頂
 * 伙伴【Thread-3】到達(dá)山頂
 * 伙伴【Thread-3】中途山腰返回
 * 伙伴【Thread-1】到達(dá)山頂
 * 大家都到山頂了，開(kāi)始休息，人數(shù)：5
 * 伙伴【Thread-5】開(kāi)始休息
 * 伙伴【Thread-2】開(kāi)始休息
 * 伙伴【Thread-6】開(kāi)始休息
 * 伙伴【Thread-0】開(kāi)始休息
 * 伙伴【Thread-1】開(kāi)始休息
 * 伙伴【Thread-2】休息結(jié)束，下山回家
 * 伙伴【Thread-0】休息結(jié)束，下山回家
 * 伙伴【Thread-1】休息結(jié)束，下山回家
 * 伙伴【Thread-5】休息結(jié)束，下山回家
 * 伙伴【Thread-6】休息結(jié)束，下山回家
 */