Java 并发编程之 Phaser

作用

可重用的同步屏障，用法与CountDownLatch和CyclicBarrier相似，支持更灵活的用法。

构造函数

parties：需要参与协作的线程数量。

parent：给定父类Phaser。

public Phaser() {
    this(null, 0);
}

public Phaser(int parties) {
    this(null, parties);
}

public Phaser(Phaser parent) {

}

方法

register()

注册一个需要协作的线程。

bulkRegister(int parties)

批量注册需要协作的线程。

arrive()

到达屏障直接执行，无需等待其他线程。

arriveAndAwaitAdvance()

到达屏障，必须等待其他线程。

arriveAndDeregister()

到达屏障，注销自己，无需等待其他线程到达。

onAdvance(int phase, int registeredParties)

参与协作的线程都到达屏障后，会调用该方法。

案例

private static void method1() {
    Phaser phaser = new Phaser() {
        @Override
        protected boolean onAdvance(int phase, int registeredParties) {
            System.out.println(LocalDateTime.now() + ": onAdvance，registeredParties=" + getRegisteredParties() + ", phase=" + getPhase() + ", isTerminated=" + isTerminated() + ", ThreadId=" + Thread.currentThread().getId());
            return super.onAdvance(phase, registeredParties);
        }
    };
    System.out.println(LocalDateTime.now() + ": 主线程开始执行异步任务，registeredParties=" + phaser.getRegisteredParties() + ", phase=" + phaser.getPhase() + ", isTerminated=" + phaser.isTerminated() + ", ThreadId=" + Thread.currentThread().getId());
    phaser.register();
    for (int i = 0; i < 5; i++) {
        phaser.register();
        System.out.println(LocalDateTime.now() + ": 注册一个屏障，registeredParties=" + phaser.getRegisteredParties() + ", phase=" + phaser.getPhase() + ", isTerminated=" + phaser.isTerminated() + ", ThreadId=" + Thread.currentThread().getId());
        int finalI = i;
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(finalI);
                System.out.println(LocalDateTime.now() + ": 到达屏障，等待其他线程，" + finalI + ", registeredParties=" + phaser.getRegisteredParties() + ", phase=" + phaser.getPhase() + ", isTerminated=" + phaser.isTerminated() + ", ThreadId=" + Thread.currentThread().getId());
                phaser.arriveAndAwaitAdvance();
                TimeUnit.SECONDS.sleep(finalI);
                System.out.println(LocalDateTime.now() + ": 屏障打开，开始执行剩下任务，" + finalI + ", registeredParties=" + phaser.getRegisteredParties() + ", phase=" + phaser.getPhase() + ", isTerminated=" + phaser.isTerminated() + ", ThreadId=" + Thread.currentThread().getId());
                phaser.arriveAndDeregister();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
    phaser.arriveAndDeregister();
    System.out.println(LocalDateTime.now() + ": 主线程执行完毕，registeredParties=" + phaser.getRegisteredParties() + ", phase=" + phaser.getPhase() + ", isTerminated=" + phaser.isTerminated() + ", ThreadId=" + Thread.currentThread().getId());
}

输出：

2020-01-22T16:15:24.545: 主线程开始执行异步任务，registeredParties=0, phase=0, isTerminated=false, ThreadId=1
2020-01-22T16:15:24.546: 注册一个屏障，registeredParties=2, phase=0, isTerminated=false, ThreadId=1
2020-01-22T16:15:24.552: 注册一个屏障，registeredParties=3, phase=0, isTerminated=false, ThreadId=1
2020-01-22T16:15:24.552: 注册一个屏障，registeredParties=4, phase=0, isTerminated=false, ThreadId=1
2020-01-22T16:15:24.552: 注册一个屏障，registeredParties=5, phase=0, isTerminated=false, ThreadId=1
2020-01-22T16:15:24.552: 注册一个屏障，registeredParties=6, phase=0, isTerminated=false, ThreadId=1
2020-01-22T16:15:24.553: 主线程执行完毕，registeredParties=5, phase=0, isTerminated=false, ThreadId=1
2020-01-22T16:15:24.553: 到达屏障，等待其他线程，0, registeredParties=5, phase=0, isTerminated=false, ThreadId=12
2020-01-22T16:15:25.553: 到达屏障，等待其他线程，1, registeredParties=5, phase=0, isTerminated=false, ThreadId=13
2020-01-22T16:15:26.554: 到达屏障，等待其他线程，2, registeredParties=5, phase=0, isTerminated=false, ThreadId=14
2020-01-22T16:15:27.553: 到达屏障，等待其他线程，3, registeredParties=5, phase=0, isTerminated=false, ThreadId=15
2020-01-22T16:15:28.553: 到达屏障，等待其他线程，4, registeredParties=5, phase=0, isTerminated=false, ThreadId=16
2020-01-22T16:15:28.553: onAdvance，registeredParties=5, phase=0, isTerminated=false, ThreadId=16
2020-01-22T16:15:28.554: 屏障打开，开始执行剩下任务，0, registeredParties=5, phase=1, isTerminated=false, ThreadId=12
2020-01-22T16:15:29.553: 屏障打开，开始执行剩下任务，1, registeredParties=4, phase=1, isTerminated=false, ThreadId=13
2020-01-22T16:15:30.554: 屏障打开，开始执行剩下任务，2, registeredParties=3, phase=1, isTerminated=false, ThreadId=14
2020-01-22T16:15:31.553: 屏障打开，开始执行剩下任务，3, registeredParties=2, phase=1, isTerminated=false, ThreadId=15
2020-01-22T16:15:32.554: 屏障打开，开始执行剩下任务，4, registeredParties=1, phase=1, isTerminated=false, ThreadId=16
2020-01-22T16:15:32.554: onAdvance，registeredParties=0, phase=1, isTerminated=false, ThreadId=16