CyclicBarrier之多线程中的循环栅栏详解

//参数parties：表示要到达屏障 (栅栏)的线程数量
//参数Runnable： 最后一个线程到达屏障之后要做的任务
​
//构造方法1
public CyclicBarrier(int parties) 
//构造方法2
public CyclicBarrier(int parties, Runnable barrierAction)
​
//线程调用await()方法表示当前线程已经到达栅栏，然后会被阻塞
public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}
​
//带时限的阻塞等待
public int await(long timeout, TimeUnit unit) throws InterruptedException,BrokenBarrierException,TimeoutException {
    return dowait(true, unit.toNanos(timeout));
}

//循环栅栏-可多次循环使用
   CyclicBarrier cyclicBarrier = new CyclicBarrier(5,()->{
        System.out.println(Thread.currentThread().getName()+" 完成最后任务！");
   });
​
   IntStream.range(1,6).forEach(i->{
        new Thread(()->{
            try {
                Thread.sleep(new Double(Math.random()*3000).longValue());
                System.out.println(Thread.currentThread().getName()+" 到达栅栏A");
                
                cyclicBarrier.await();//屏障点A，当前线程会阻塞至此，等待计数器=0
                
                System.out.println(Thread.currentThread().getName()+" 冲破栅栏A");
            }catch (Exception e){
                e.printStackTrace();
            }
        }).start();
    });

 /** CyclicBarrier使用的排他锁*/
    private final ReentrantLock lock = new ReentrantLock();
    /** barrier被冲破前，线程等待的condition*/
    private final Condition trip = lock.newCondition();
    /** barrier被冲破时，需要满足的参与线程数。*/
    private final int parties;
    /* barrier被冲破后执行的方法。*/
    private final Runnable barrierCommand;
    /** 当其轮次 */
    private Generation generation = new Generation();
​
    /**
     *目前等待剩余的参与者数量。从 parties倒数到0。每个轮次该值会被重置回parties
     */
    private int count;

//构造器1：指定本局要拦截的线程数parties 及 本局结束时要执行的任务
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}
​
//构造器2
public CyclicBarrier(int parties) {
    this(parties, null);
}

private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException,TimeoutException {
    // 获取独占锁
    final ReentrantLock lock = this.lock;
    lock.lock();//对共享资源count,generation操作前，需先上锁保证线程安全
    try {
        // 当前代--当前轮次对象的引用
        final Generation g = generation;
        // 如果这轮次broken了，抛出异常
        if (g.broken)
            throw new BrokenBarrierException();
        // 如果线程中断了，抛出异常
        if (Thread.interrupted()) {
            breakBarrier();//如果被打断，通过此方法设置当前轮次为broken状态，通知当前轮次所有等待的线程
            throw new InterruptedException();//抛出异常
        }
 
        //自旋前
        
        //1、count值-1
        int index = --count;
        // 2、判断是否到0，若是，则冲破屏障点（说明最后一个线程已经到达）
        if (index == 0) {
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                // 3、执行栅栏任务（若CyclicBarrier构造时传入了Runnable，则调用）
                if (command != null)
                    command.run();
                ranAction = true;
                // 4、更新一轮次，将count重置，将generation重置，唤醒之前等待的线程
                nextGeneration();
                return 0;
            } finally {
                // 如果执行栅栏任务（command）的时候出现了异常，那么就认为本轮次破环了
                if (!ranAction)
                    breakBarrier();
            }
        }
 
        //计数器没有到0 =》开始自旋，直到屏障被冲破，或者interrupted或者超时
        for (;;) {
            try {
                 // 开始等待；如果没有时间限制，则直接等待，直到被唤醒（让其他线程进入到lock的代码块执行以上逻辑）
                if (!timed)
                    trip.await();//阻塞，此时会释放锁，以让其他线程进入await方法中，等待屏障被冲破后，向后执行
                // 如果有时间限制，则等待指定时间
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                // 如果当前线程阻塞被interrupt了，并且本轮次还没有被break，那么修改本轮次状态为broken
                if (g == generation && ! g.broken) {
                    // 让栅栏失效
                    breakBarrier();
                    throw ie;
                } else {
                    // 上面条件不满足，说明这个线程不是这轮次的，就不会影响当前这代栅栏的执行，所以，就打个中断标记
                    Thread.currentThread().interrupt();
                }
            }
 
            // 当有任何一个线程中断了，就会调用breakBarrier方法,就会唤醒其他的线程，其他线程醒来后，也要抛出异常 
            if (g.broken)
                throw new BrokenBarrierException();
 
            // g != generation表示正常换轮次了，返回当前线程所在栅栏的下标
            // 如果 g == generation，说明还没有换，那为什么会醒了？
            // 因为一个线程可以使用多个栅栏，当别的栅栏唤醒了这个线程，就会走到这里，所以需要判断是否是当前代。
            // 正是因为这个原因，才需要generation来保证正确。
            if (g != generation)
                return index;
            
            // 如果有时间限制，且时间小于等于0，销毁栅栏并抛出异常
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }//自旋
        
    } finally {
        // 释放独占锁
        lock.unlock();
    }
}

private void nextGeneration() {
    trip.signalAll();//唤醒本轮次等待的线程
    count = parties;//重置count值为初始值，为下一轮次（代）使用
    generation = new Generation();//更新本轮次对象，这样自旋中的线程才会跳出自旋。
}
​
private static class Generation {
    boolean broken = false;
}
​
private void breakBarrier() {
    generation.broken = true;//设置标识
    count = parties;//重置count值为初始值
    trip.signalAll();//唤醒所有等待线程
}