Java7:理解Phaser

转自 http://hrps.me/2016/11/22/java-concurrent-phaser/

Java7引入一种称为Phaser的灵活的线程同步机制。如果你需要所有线程到达之后，才继续或者开始进行新的一组任务，Phaser是一个很好的选择。
下面是代码，及每一步的解释。
感觉原代码解释的不好，我从新写了个感觉更好理解。😁

import java.util.concurrent.Phaser;

/**
 * Description:
 * User: Huang rp
 * Date: 16/11/23
 * Version: 1.0
 */
public class PhaserSamples {

    public static void main(String[] args) throws InterruptedException {
        new PhaserSamples().runTasks();

    }

    // 一共开几场会
    private static final int PHASE_TO_TERMINATE = 2;
    // 初始参会人员数量
    private static final int INIT_PARTIES = 1;
    // 增加参会人员数量
    private static final int ADD_PARTIES = 5;
    // 每个会场限制参会者数量
    private static final int TASKS_PER_PHASER = 10;

    void runTasks() throws InterruptedException {
        final Phaser phaser = new Phaser(INIT_PARTIES) {
            // 所有人员到达完毕，开始会议；连续开PHASE_TO_TERMINATE - 1 场会议，会议结束（terminal）
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("第" + (phase + 1) + "场会议结束");
                return phase == (PHASE_TO_TERMINATE - 1) || registeredParties == 0;
            }
        };
        final Task tasks[] = new Task[ADD_PARTIES];

        System.out.println("会议开始准备,需要至少" + INIT_PARTIES + "人签到, 共连续" + PHASE_TO_TERMINATE + "场会议,分会场数量" +
                (ADD_PARTIES % TASKS_PER_PHASER == 0 ? (ADD_PARTIES / TASKS_PER_PHASER) : (ADD_PARTIES / TASKS_PER_PHASER + 1)));
        System.out.println("Main准备参加第一场会议,开始签到,共" + phaser.getRegisteredParties() + "人签到");

        build(tasks, 0, tasks.length, phaser);
        for (int i = 0; i < tasks.length; i++) {
            final Thread thread = new Thread(tasks[i]);
            thread.start();
        }

        phaser.arriveAndDeregister();
        System.out.println("Main离开会场");
        // 准备下一场会议，如果没有再到达会场，所有人都将等待
//        phaser.arriveAndAwaitAdvance();
//        System.out.println("Main休息准备下一场会议");
    }

    public static void build(Task[] tasks, int lo, int hi, Phaser ph) {
        if (hi - lo > TASKS_PER_PHASER) {
            for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
                int j = Math.min(i + TASKS_PER_PHASER, hi);
                build(tasks, i, j, new Phaser(ph));
            }
        } else {
            for (int i = lo; i < hi; ++i)
                tasks[i] = new Task(i + 1, ph);
        }
    }


    public static class Task implements Runnable {
        private final int id;
        private final Phaser phaser;

        public Task(int id, Phaser phaser) {
            this.id = id;
            this.phaser = phaser;
            this.phaser.register();
            System.out.println("参会人员" + id + "已经签到,共" + phaser.getRegisteredParties() + "人签到");
        }

        @Override
        public void run() {
            // 参加每一场会议
            while (!phaser.isTerminated()) {
                try {
                    // 签到之后，走到会场
                    Thread.sleep(20 * id + 20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("参会人员" + this.id + "已经到达第" + (phaser.getPhase() + 1) + "场会议, 当前共有"
                        + phaser.getRegisteredParties() + " 人参会,"
                        + (phaser.getArrivedParties() + 1) + " 到达,"
                        + (phaser.getUnarrivedParties() - 1) + " 未到达");
                // 到达会场
                phaser.arriveAndAwaitAdvance();
            }
        }
    }
}

运行结果

会议开始准备,需要至少1人签到, 共连续2场会议,分会场数量1
Main准备参加第一场会议,开始签到,共1人签到
参会人员1已经签到,共2人签到
参会人员2已经签到,共3人签到
参会人员3已经签到,共4人签到
参会人员4已经签到,共5人签到
参会人员5已经签到,共6人签到
Main离开会场
参会人员1已经到达第1场会议, 当前共有5 人参会,1 到达,4 未到达
参会人员2已经到达第1场会议, 当前共有5 人参会,2 到达,3 未到达
参会人员3已经到达第1场会议, 当前共有5 人参会,3 到达,2 未到达
参会人员4已经到达第1场会议, 当前共有5 人参会,4 到达,1 未到达
参会人员5已经到达第1场会议, 当前共有5 人参会,5 到达,0 未到达
第1场会议结束
参会人员1已经到达第2场会议, 当前共有5 人参会,1 到达,4 未到达
参会人员2已经到达第2场会议, 当前共有5 人参会,2 到达,3 未到达
参会人员3已经到达第2场会议, 当前共有5 人参会,3 到达,2 未到达
参会人员4已经到达第2场会议, 当前共有5 人参会,4 到达,1 未到达
参会人员5已经到达第2场会议, 当前共有5 人参会,5 到达,0 未到达
第2场会议结束

有几点注意：
当签到的人员数量等于到达会场人员数量，将立即开会，所以第一个人员到达会场会比最后一个签到的晚才能保证所有人能参加会议。
可以一个会场容纳所有人开会，将同一个phaser传递给所有的task；也可以维护一个层级关系，建立多个会场进行同一个会议，将所有参会人员平均分配到各个会场。
签到之后，可以不用等待所有人到达会场可以立即离场，并且从签到簿上抹掉签名。
下一场会议开始之前，如果还有人员等待下一场会议，那么所有签到的人都必须参加下一场。如果所有人都不等待，会议结束。
所有人员参加完指定数量的会议后，会议结束。

Phaser终止的两种途径，Phaser维护的线程执行完毕或者onAdvance()返回true
此外Phaser还能维护一个树状的层级关系，构造的时候new Phaser(parentPhaser)，对于Task执行时间短的场景（竞争激烈），** TASKS_PER_PHASER**值设置较小，反之适当增大。

名词解释：

• party 对应一个线程，数量可以通过register或者初始化new Phaser(num)的时候控制
• arrive 对应一个party的状态，初始时是unarrived，当调用arriveAndAwaitAdvance()或者 arriveAndDeregister()进入arrive状态，可以通过getUnarrivedParties()获取当前未到达的数量
• register 注册一个party，每一阶段必须所有注册的party都到达才能进入下一阶段
• phase 阶段，当所有注册的party都arrive之后，将会调用Phaser的onAdvance()方法来判断是否要进入下一阶段