CountDownLatch与CyclicBarrier对比

在并发编程时总会遇到一种这样的场景：等待一系列任务做完后，才能开始做某个任务。当遇到这种场景时，两个类cross our mind:CountDownLatch和CyclicBarrier。下面从使用方法和内部实现原理分别对这两个类做出介绍。

使用方法

CountDownLatch

任务

class MyThread extends Thread{

    private CountDownLatch latch;

    public MyThread(CountDownLatch latch) {
        this.latch = latch;
    }

    @Override
    public void run() {
        try {
            System.out.println(Thread.currentThread().getName());
            Thread.sleep(100);
            // 任务完成 state - 1
            latch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            latch.countDown();
        }
    }
}

在完成每一个任务后，latch中的int数字做减一操作。

测试

public class CountDownLatchTest {
    @Test
    public void main() {
        // 初始化值为3
        CountDownLatch latch = new CountDownLatch(3);
        // 启动3个任务
        for (int i = 3; i > 0; i --) {
            new MyThread(latch).start();
        }
        try {
            // 等待三个任务完成
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("count = " + latch.getCount());
        System.out.println("finished");
    }
}

主线程中启动了三个子线程，然后调用了latch.await()方法。

输出结果

Thread-1
Thread-0
Thread-2
count = 0
finished

从输出结果可以看出，主线程在等待三个子线程完成任务之后才结束的。

CyclicBarrier

先完成的任务

public class NormalTask implements Runnable {
    CyclicBarrier barrier;

    NormalTask(CyclicBarrier barrier) {
        this.barrier = barrier;
    }
    @Override
    public void run() {

        try {
            Thread.sleep(100);
            barrier.await();
            System.out.println(System.currentTimeMillis() + " first step finished");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

每个任务完成需要100ms。

后完成的任务

public class FinalTask implements Runnable {
    @Override
    public void run() {
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(System.currentTimeMillis() + " second step finished");
    }
}

后完成的任务需要10ms。

主线程

主线程启动了两个先执行的线程，将后完成的线程作为参数传入CyclicBarrier。

@Test
public void testInterruptException() throws InterruptedException {
    // 主线程作为参数传入，主线程需要等待子线程完成
    CyclicBarrier barrier = new CyclicBarrier(2,new FinalTask());
    new Thread(new NormalTask(barrier)).start();
    new Thread(new NormalTask(barrier)).start();
    Thread.sleep(300);
}

运行结果

1543326017854 first step finished
1543326017854 first step finished
1543326017870 second step finished

从运行结构可以看出，先启动的任务几乎同时完成，而后完成的任务结束时间比前两个线程完成时间晚16ms,其中6ms是启动线程所花费的。主线程中sleep 300ms 是为了等待所有的线程都执行完成。也可以使用join实现相同的效果。在这里解释一下为什么不能像CountDownLatch一样用主线程作为等待线程。我刚开始也是这样做的，发现主线程一下就跑完了，根本不停。查看了源码才发现，CyclicBarrier没有park主线程。具体逻辑相见下文的原理分析。

相同点

两个类都可以实现一个任务等待其他几个任务完成后再执行。

不同点

• 任务中调用的接口不同：CountDownLatch在任务完成后调用的是countDown函数。在等待线程中调用了await方法。而CyclicBarrier只在先开始的任务中调用了await方法。后运行任务中没有涉及到任何和CyclicBarrier的信息。
• CountDownLatch 在完成所有的操作后不可重用了，但是CyclicBarrier可以在完成任务或者有线程抛出异常后调用reset方法继续使用，这应该是这个类叫做循环屏障的原因。

原理

两个类都是在初始化时，传入一个整形数字，表示需要等待几个任务完成后才能开始执行等待的任务。但是其底层实现的原理完全不同。下面对两个类的实现原理做具体介绍。

CountDownLatch

CountDownLatch park 的是主线程，是主线程和所有的子线程在竞争同一把锁。但是初始化时，他把锁默认给了子线程（将AQS中的state 置为需要等待的子线程的个数）。

Sync(int count) {
    setState(count);
}
public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

而主线程在调用await方法时，先检测state是否为0，如果=0 就不用park了，这时说明子线程都已完成了。如果!= 0。则park。
每个子线程在执行完任务后，将state使用cas的方式减1，并尝试取唤醒(unpark)主线程。

public void countDown() {
    sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}
// 通过cas将state减1 如果state = 0 则调用doReleaseShared唤醒AQS队列中的主线程
protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

上面完整的介绍了CountDownLatch的工作原理。

CyclicBarrier

为了与CountDwonLatch 对比，也为了方便描述问题，我们将先执行的任务叫做子线程，将后执行的任务叫做主线程。
CyclicBarrier 在初始化时将int值不但赋值给了state,其内部也留了一个备份，这就是CyclicBarrier可以调用reset重新使用的一个原因。而且其内部是在可重入锁ReentrantLock和Condition的基础上实现的，在其代码内部几乎看不到CAS代码，看到更多的是重入锁的lock和unlock以及Condition的await和singal。

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

其中的parties就是子线程个数的备份，而barrierAction可有可无。
在子任务完成后就会调用await方法：

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

其核心逻辑在dowait方法中。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();
        }
        ....

如果没有线程抛出异常，将count减一，并检查count是否为0 如果不为0 将当前的线程放入Condition的等待队列。如果等于0 则唤醒之前的所有线程。

 int index = --count;
    if (index == 0) {  // 如果等于0说明所有的任务都已完成，唤醒所有Condition中的线程。
        boolean ranAction = false;
        try {
            final Runnable command = barrierCommand;
            if (command != null)
                command.run();
            ranAction = true;
            nextGeneration();
            return 0;
        } finally {
            if (!ranAction)
                breakBarrier();
        }
    }

    // loop until tripped, broken, interrupted, or timed out
    for (;;) {
        try {
            if (!timed)
                trip.await(); // 放入Condition队列中
            else if (nanos > 0L)
                nanos = trip.awaitNanos(nanos);
        } catch (InterruptedException ie) {

至此，CyclicBarrier的原理页介绍完成了。

不同点

• CountDownLatch 在AQS队列中park的是主线程，而CyclicBarrier在AQS中park的是所有的子线程。
• CountDownLatch 是放到AQS队列中，而CyclicBarrier是将子线程放到Condition队列中。
• CountDownLatch 唤醒的是主线程，而CyclicBarrier 是通过singleAll函数，将所有的子线程移动到AQS队列中，然后再开始执行。

总结

通过以上分析可以得出，CountDownLatch 更适合一个任务等待一些任务执行完成后再执行，而CyclicBarrier更适合保证一批任务同时结束。