Java中的CyclicBarrier循环栅栏深入解析

public class Demo1 {
    public static void main(String[] args) {
        // 初始化一个 CyclicBarrier，大小为 5，当 5 个线程满足条件之后，就执行一次：执行完毕！
        CyclicBarrier cyclicBarrier = new CyclicBarrier( 5, () -> {
            System.out.println("执行完毕！");
        });
        // 为了测试是否能循环，这里循环设置为 20，预期的结果应该是会执行 4 次：执行完毕！
        for (int i=0; i<20; i++) {
            int temp = i;
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + ", 执行到：" + temp );
                try{
                    cyclicBarrier.await();  // 等待其他线程到齐，到齐了之后就会执行一次
                } catch (BrokenBarrierException | InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

//同步操作锁
private final ReentrantLock lock = new ReentrantLock();
//线程拦截器
private final Condition trip = lock.newCondition();
//每次拦截的线程数
private final int parties;
//换代前执行的任务
private final Runnable barrierCommand;
//表示栅栏的当前代
private Generation generation = new Generation();
//计数器
private int count;
//静态内部类Generation
private static class Generation {
    Generation() {}                 // 防止访问构造函数创建
    boolean broken;                 // 初始化为 false
}

//构造器，这里就是我们上面那个例子使用的构造器
public CyclicBarrier(int parties, Runnable barrierAction) {
  if (parties <= 0) throw new IllegalArgumentException();
  this.parties = parties;
  this.count = parties;
  this.barrierCommand = barrierAction;
}
// 构造器，只传入一个 parties 的大小，不进行自定义任务
public CyclicBarrier(int parties) {
  this(parties, null);
}

// 非定时等待
public int await() throws InterruptedException, BrokenBarrierException {
  try {
    return dowait(false, 0L);
  } catch (TimeoutException toe) {
    throw new Error(toe);
  }
}
// 定时等待
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException {
  return dowait(true, unit.toNanos(timeout));
}

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()) {
            // 如果中断了，就会执行下面三件事
            // 1. 将 generation 的 broken 标记设置为 true
            // 2. 重置 count 值
            // 3. 唤醒当前拦截的线程
            breakBarrier();
            throw new InterruptedException();
        }
        int index = --count;    // 每次都将计数器的值减 1
        if (index == 0) {  // // 计数器的值减为0，就会唤醒所有线程并转换到下一代
            Runnable command = barrierCommand;
            if (command != null) {    // // 如果存在自定义的任务，这里就会做判断，在唤醒所有线程之前执行
                try {
                    command.run();
                } catch (Throwable ex) {
                    breakBarrier();
                    throw ex;
                }
            }
            nextGeneration();    // 唤醒所有线程，并转到下一代
            return 0;
        }
        //官方注释： loop until tripped, broken, interrupted, or timed out
        //这里循环是：如果计数器不为0则执行此循环
        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 {
                     //若在捕获中断异常前已经完成在栅栏上的等待, 则直接调用中断操作
                    Thread.currentThread().interrupt();
                }
            }
            if (g.broken)
                throw new BrokenBarrierException();
            if (g != generation) //如果线程因为换代操作而被唤醒则返回计数器的值
                return index;
            // 如果是因为时间到了而被唤醒则打破栅栏，抛出异常
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();    // 解锁
    }
}

//切换栅栏到下一代
private void nextGeneration() {
  //唤醒条件队列所有线程
  trip.signalAll();
  //设置计数器的值为需要拦截的线程数
  count = parties;
  //重新设置栅栏代次
  generation = new Generation();
}
//打破当前栅栏
private void breakBarrier() {
  //将当前栅栏状态设置为打翻
  generation.broken = true;
  //设置计数器的值为需要拦截的线程数
  count = parties;
  //唤醒所有线程
  trip.signalAll();
}

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

public class Demo {
    static class Tourist extends Thread {
        CyclicBarrier cyclicBarrier;
        public Tourist(CyclicBarrier cyclicBarrier){
            this.cyclicBarrier = cyclicBarrier;
        }
        @Override
        public void run() {
            try {
                // 模拟点先各自独立运行
                Thread.sleep( (int) (Math.random() * 1000) );
                // 集合点 A
                cyclicBarrier.await();
                System.out.println(this.getName() + "，到达集合点 A " + System.currentTimeMillis() );
                // 集合后模拟再各自独立运行
                Thread.sleep( (int) (Math.random() * 1000));
                // 集合点 B
                cyclicBarrier.await();
                System.out.println(this.getName() + "，到达集合点 B " + System.currentTimeMillis() );
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        // 定义出多个 线程
        int num = 3;
        Tourist[] tourist = new Tourist[num];
        CyclicBarrier barrier = new CyclicBarrier(3, () -> {
            System.out.println("集合完毕： " + System.currentTimeMillis() + " 执行的线程为： " + Thread.currentThread().getName());
        });
        for (int i = 0; i< num; i++){
            tourist[i] = new Tourist(barrier);
            tourist[i].start();
        }
    }
}