CyclicBarrier

一.使用

一组线程相互等待，当所有线程都到达某个屏障点后再进行后续的操作。

public class CyclicBarrier1 {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                Thread.sleep(2000);
                System.out.println(Thread.currentThread().getName() + "-----------------------------");
            }
        });
        for (int i = 0; i < 5; i++) {
            new MyThread(cyclicBarrier).start();
        }
    }
}
class MyThread extends Thread {
    CyclicBarrier cyclicBarrier;
    public MyThread(CyclicBarrier cyclicBarrier) {
        this.cyclicBarrier = cyclicBarrier;
    }
    @SneakyThrows
    @Override
    public void run() {
        System.out.println(getName() + "await1");
        cyclicBarrier.await();
        System.out.println(getName() + "await2");
        cyclicBarrier.await();
        System.out.println(getName() + "end");
    }
}
输出：
Thread-0await1
Thread-1await1
Thread-2await1
Thread-3await1
Thread-4await1
Thread-4-----------------------------
Thread-4await2
Thread-0await2
Thread-2await2
Thread-1await2
Thread-3await2
Thread-3-----------------------------
Thread-3end
Thread-4end
Thread-0end
Thread-1end
Thread-2end

二.源码

1.构造方法

private final ReentrantLock lock = new ReentrantLock();
private final Condition trip = lock.newCondition();
public CyclicBarrier(int parties, Runnable barrierAction) {
    // 每次拦截的线程数
    this.parties = parties;
    // 计数器
    this.count = parties;
    // 换代前执行的任务
    this.barrierCommand = barrierAction;
}

2.await

public int await() throws InterruptedException, BrokenBarrierException {
    return dowait(false, 0L);
}
private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException, TimeoutException {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 栅栏的当前代
        final Generation g = generation;
        int index = --count;
        // 最后1个线程到达
        if (index == 0) {
            final Runnable command = barrierCommand;
            if (command != null)
                command.run();
            //切换栅栏到下一代
            nextGeneration();
            return 0;
        }
        for (;;) {
            if (!timed)
                // 阻塞
                trip.await();
            if (g != generation)
                return index;
        }
    } finally {
        lock.unlock();
    }
}
private void nextGeneration() {
    // 唤醒所有线程
    trip.signalAll();
    count = parties;   
    generation = new Generation();
}