JAVA并发编程系列(9)CyclicBarrier循环屏障原理分析

package lading.java.mutithread;
import cn.hutool.core.date.DateTime;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
/**
 * 模拟拼团，并通知仓库发货
 */
public class Demo010CyclicBarrier {
    public static int count = 2;//要求达到屏障数量
    public static volatile ConcurrentHashMap<String, String> customerNames = new ConcurrentHashMap<>();//记录已到达屏障客户线程名
    public static CyclicBarrier barrier = new CyclicBarrier(count, new sendMsg());//屏障数量2，目标线程数量达到后，执行sendMsg
    public static void main(String[] args) {
        //模拟5个人拼团
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    Thread.sleep((new Random().nextInt(10 - 1 + 1)) * 1000);//客户浏览商品信息Ns
                    System.out.println(DateTime.now().toString("YYYY-MM-dd hh:mm:ss SSS") +" 【"+ Thread.currentThread().getName() + "】,到达屏障");
                    customerNames.put(Thread.currentThread().getName(), Thread.currentThread().getName());//本批次拼团名单
                    barrier.await();//到达屏障，进入阻塞
                    System.out.println(DateTime.now().toString("YYYY-MM-dd hh:mm:ss SSS") +" 【"+ Thread.currentThread().getName() + "】,完成支付。");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (BrokenBarrierException e) {
                    throw new RuntimeException(e);
                }
            }, "客户00" + (i + 1)).start();
        }
    }
    static class sendMsg extends Thread {
        @Override
        public void run() {
            System.out.println(DateTime.now().toString("YYYY-MM-dd hh:mm:ss") + customerNames.keySet() + "达到屏障拼团人数，允许提交订单！");
            customerNames.clear();//本批次拼团名单清空
        }
    }
}

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;

public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }
    //不指定屏障任务
    public CyclicBarrier(int parties) {
        this(parties, null);
    }

private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        //加锁，去更新count，这里就不是CAS了
        lock.lock();
        try {
            final Generation g = generation;
            //判断屏障是否被中断
            if (g.broken)
                throw new BrokenBarrierException();
            //判断本线程是否已中断
            if (Thread.interrupted()) {
                //本组线程的屏障中断，并重置屏障
                breakBarrier();
                throw new InterruptedException();
            }
            // 对count减1
            int index = --count;
            // 本组屏障全部线程都到达屏障，接下来执行屏障任务、以及唤醒本组其他阻塞兄弟
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    //实例CyclicBarrier(),如果有指定任务，本线程就代劳去执行
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
            // 如果不为0，说明约定到本屏障的其他兄弟们还没到齐，那就自旋等待，直到被打断或者超时
            for (;;) {
                try {
                    if (!timed)
                        trip.await();
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } 
      ........
    }