引言

这篇文章，我们来看CyclicBarrier。CyclicBarrier可以让多个线程都运行到同一点，在所有的线程都到了该点之后再继续执行。CyclicBarrier内部的实现用到了Lock和Condition。

一个例子

public class CyclicBarrierTest {
    public static void main(String[] args) throws InterruptedException {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(8, new Runnable() {
            @Override
            public void run() {
                System.out.println("线程"+Thread.currentThread().getName()+"最后完成任务");
            }
        });
        for(int i=0;i<8;i++){
            Thread thread = new Thread(new Task(cyclicBarrier),"thread_"+i);
            thread.start();
        }
    }
    static class Task implements Runnable{
        public Task(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }
        private CyclicBarrier cyclicBarrier;
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("线程"+Thread.currentThread().getName()+"到达栅栏A");
            try {
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println("线程"+Thread.currentThread().getName()+"离开栅栏A");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("线程"+Thread.currentThread().getName()+"到达栅栏B");
            try {
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println("线程"+Thread.currentThread().getName()+"离开栅栏B");
        }
    }
}

在这个例子中，cyclicBarrier设置了两个栅栏，所有的线程先到达栅栏A，然后再到达栅栏B。运行结果如下：

线程thread_4到达栅栏A
线程thread_1到达栅栏A
线程thread_3到达栅栏A
线程thread_5到达栅栏A
线程thread_2到达栅栏A
线程thread_7到达栅栏A
线程thread_6到达栅栏A
线程thread_0到达栅栏A
线程thread_0最后完成任务
线程thread_0离开栅栏A
线程thread_4离开栅栏A
线程thread_3离开栅栏A
线程thread_1离开栅栏A
线程thread_5离开栅栏A
线程thread_7离开栅栏A
线程thread_6离开栅栏A
线程thread_2离开栅栏A
线程thread_5到达栅栏B
线程thread_4到达栅栏B
线程thread_7到达栅栏B
线程thread_0到达栅栏B
线程thread_3到达栅栏B
线程thread_6到达栅栏B
线程thread_2到达栅栏B
线程thread_1到达栅栏B
线程thread_1最后完成任务
线程thread_1离开栅栏B
线程thread_5离开栅栏B
线程thread_0离开栅栏B
线程thread_7离开栅栏B
线程thread_4离开栅栏B
线程thread_2离开栅栏B
线程thread_6离开栅栏B
线程thread_3离开栅栏B

实现分析

构造方法

public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

这个是我们上面例子中用到的构造方法，它初始化了CyclicBarrier中几个重要的字段，这里我先列出来CyclicBarrier所有的字段：

/** 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 */
    private final Runnable barrierCommand;
    /** The current generation */
    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.
     */
    private int count;

前两个是重入锁和Condition，它俩用来实现线程等待和唤醒的逻辑，这个我们稍后会看到。parties可以认为是等待的线程继续执行前需要执行await方法的次数，对于同一个CyclicBarrier，它设置的每个屏障的parties都是一样的。count是在一个屏障中，仍然处于等待状态的数量，所以初始化CyclicBarrier时，count=parties，但是之后count会发生变化而parties不会。barrierCommand用来指定最后到达屏障的线程要执行的命令。generation用来分代，因为一个CyclicBarrier可以重复使用，每次使用都认为是新的一代。

await方法

await方法的实现如下：

public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

看dowait：

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();
            }
            int index = --count;
            //index=0说明最后一个线程执行await方法
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    //如果command不为空 执行
                    if (command != null)
                        command.run();
                    ranAction = true;
                    //新的分代 也就是重新设置屏障 nextGeneration方法中会执行signalAll方法唤醒所有等待的线程
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
            //如果count不等于0 说明当前线程不是最后执行await方法的线程   
            // 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)
                    return index;
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

首先，它会使用lock进行加锁，也就是await方法是互斥执行的。然后判断当前线程是否是最后执行await方法的线程，如果是，它会首先执行构造方法中传入的Runnable，然后调用nextGeneration方法，该方法很关键：

private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation();
    }

首先，它会使用Condition的signalAll来唤醒所有等待的线程，然后重置count。
如果当前线程不是最后执行await的线程，它就会调用Condition的await方法来等待。
所以基于lock、condition，await实现了最后一个线程到达屏障点，然后唤醒所有之前等待的线程的逻辑。
CyclicBarrier还提供了其他很有用的方法，这里不再一一介绍。

小结

CyclicBarrier能够实现CountDownLatch的功能，同时也提供了CountDownLatch没有的功能，例如多次设置屏障等，并且它俩的实现原理有很大差别，CountDownLatch基于AQS共享模式，CyclicBarrier基于Lock和Condition。根据场景的不同，我们需要合理的选择这两个工具类来达到目的。