CyclicBarrier源码分析

 public class CyclicBarrierTest {

    public static void main(String[] args) {
        CyclicBarrierTest cyclicBarrierTest=new CyclicBarrierTest();
        cyclicBarrierTest.runThread();
    }
    //有10个线程需要相互等待
    CyclicBarrier cyclicBarrier=new CyclicBarrier(10);

    /**
     * 创建一个线程
     * @return
     */
    private Thread createThread(int i){
        Thread thread=new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                
                //线程在此相互等待，直到所有线程都准备好
                    cyclicBarrier.await();
                    System.out.println("thread"+Thread.currentThread().getName()+"准备完毕"+System.currentTimeMillis());
                }catch (InterruptedException e){
                    e.printStackTrace();
                }catch (BrokenBarrierException e){
                    e.printStackTrace();
                }

            }
        });
        thread.setName("thread-"+i);
        return  thread;
    }

    public void runThread(){
        ExecutorService executorService= Executors.newFixedThreadPool(10);

        try {
            for(int i=0;i<10;i++){
                Thread.sleep(100);
                executorService.submit(createThread(i));
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }

    }
}

private static class Generation {
    //是否被销毁
    boolean broken = false;//false代表没被销毁
}

/** The lock for guarding barrier entry ,守护入口的锁*/
private final ReentrantLock lock = new ReentrantLock();
/** Condition to wait on until tripped，等待条件 */
private final Condition trip = lock.newCondition();
/** The number of parties，要屏障的线程数 */
private final int parties;
/* The command to run when tripped ，当线程都到待barrier，需要运行的内容*/
private final Runnable barrierCommand;
/** The current generation ,记录当前barrier状态的对象*/
private Generation generation = new Generation();

/**
 * Number of parties still waiting. Counts down from parties to 0
 * on each generation.  It is reset to parties on each new
 * generation or when broken.
 *当前等待barrier到达的线程的数量
 */
private int count;

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 {
            //获取当前的generation
            final Generation g = generation;

            if (g.broken)
                throw new BrokenBarrierException();

            if (Thread.interrupted()) {
                breakBarrier(); //如果线程被中断，就会终止Barrier，唤醒所有的等待线程
                throw new InterruptedException();
            }

            //count就是我们实例化CyclicBarrier时传入的值
            //此时index代表当前是最后几个等待的线程
            int index = --count;
            if (index == 0) {  // tripped
                //如过当前线程是最后一个等待的线程
                //它都已经调用await，说明所有线程都已经到达
                //屏障点了，可以唤醒所有线程了
                boolean ranAction = false;
                try {
                //如果有barrierCommand，就运行它
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    //更新Barrier状态，并唤醒所有线程
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }

            // loop until tripped, broken, interrupted, or timed out
            //自旋等待，直到所有线程都到达屏障点
            //或者发生中断
            //或者generation被销毁
            //或者超时
            for (;;) {
                try {
                    if (!timed)
                        trip.await(); //在此利用lock的Condition阻塞，当前线程
                    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)
                    return index;

                if (timed && nanos <= 0L) {
                    breakBarrier(); //超时就销毁当前Barrier
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

private void breakBarrier() {
//将当前gengeration标记为弃用状态
    generation.broken = true;
    count = parties;//将等待barrier的线程数量，恢复到之前的值
    trip.signalAll(); //唤醒锁上的Condition上等待的所有线程
}

private void nextGeneration() {
    // signal completion of last generation
    trip.signalAll(); //唤醒所有的等待线程
    // set up next generation
    count = parties;//将等待线程数复原，以便CyclicBarrier下次重复使用
    generation = new Generation(); //复原generation
}