多线程(十二) 同步工具类-CyclicBarrier

public class Test {

    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {
        @Override
        public void run() {
            System.out.println("屏障点回调线程，执行者:" + Thread.currentThread().getName());
        }
    });

    public static void main(String[] args) {

        for(int i = 0; i< 5;i++){

            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(Thread.currentThread().getName() + "开始运行");
                        cyclicBarrier.await();
                        System.out.println(Thread.currentThread().getName() + "结束运行");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });

            thread.setName("线程" + i);
            thread.start();
        }

    }
}

// 设置屏障阈值
public CyclicBarrier(int parties) {
    this(parties, null);
}

// 设置屏障阈值，同时增加回调线程功能
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

// lock对象
private final ReentrantLock lock = new ReentrantLock();

// 跳闸，可以理解为打开屏障
private final Condition trip = lock.newCondition();

// 屏障阈值
private final int parties;

// 回调线程
private final Runnable barrierCommand;

// 每次使用屏障都会生成，内部的broken标记屏障是否破损
private Generation generation = new Generation();

// 默认设置false
private static class Generation {
    boolean broken = false;
}

    // 内部调用dowait()方法，并且参数传false，不支持超时
    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

    // 真正进入等待的逻辑
    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
    final ReentrantLock lock = this.lock;
    // 获取排他锁
    lock.lock();
    try {
        final Generation g = generation;
        // 屏障被破坏则抛异常
        if (g.broken)
            throw new BrokenBarrierException();

        if (Thread.interrupted()) {
            // 线程中断 则退出屏障
            breakBarrier();
            throw new InterruptedException();
        }

        // 到达屏障的计数-1
        int index = --count;      
        if (index == 0) {  // tripped
            // index == 0, 说明指定 count 的线程均到达屏障，此时可以打开屏障
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                if (command != null)
                    // 若指定了 barrierCommand 则执行
                    command.run();
                ranAction = true;
                // 唤醒阻塞在屏障的线程并重置 generation
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // loop until tripped, broken, interrupted, or timed out
        for (;;) {
            try {
                if (!timed)
                    // 若未指定阻塞在屏障处的等待时间，则一直等待；直至最后一个线程到达屏障处的时候被唤醒
                    trip.await();
                else if (nanos > 0L)
                    // 若指定了阻塞在屏障处的等待时间，则在指定时间到达时会返回
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                if (g == generation && ! g.broken) {
                    // 若等待过程中，线程发生了中断，则退出屏障
                    breakBarrier();
                    throw ie;
                } else {
                    // We're about to finish waiting even if we had not
                    // been interrupted, so this interrupt is deemed to
                    // "belong" to subsequent execution.
                    Thread.currentThread().interrupt();
                }
            }

            // 屏障被破坏 则抛出异常
            if (g.broken)
                throw new BrokenBarrierException();

            if (g != generation)
                // g != generation 说明所有线程均到达屏障处 可直接返回
                // 因为所有线程到达屏障处的时候，会重置 generation
                // 参考 nextGeneration
                return index;

            if (timed && nanos <= 0L) {
                // 说明指定时间内，还有线程未到达屏障处，也就是等待超时
                // 退出屏障
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}

private void nextGeneration() {

    // 唤醒阻塞在等待队列的线程
    trip.signalAll();

    // 重置 count
    count = parties;

    // 重置 generation
    generation = new Generation();
}

private void breakBarrier() {

    // broken 设置为 true
    generation.broken = true;

    // 重置 count
    count = parties;

    // 唤醒等待队列的线程
    trip.signalAll();
}

final ReentrantLock lock = this.lock;
lock.lock();
try {
    // 唤醒阻塞的线程
    breakBarrier();   // break the current generation
    // 重新设置 generation
    nextGeneration(); // start a new generation
} finally {
    lock.unlock();
}