Java并发与锁设计实现详述（7）- Java并发类CountDownLatch、CyclicBarrier、Semaphore

浅浅的花香味﹌ 2022-05-28 13:45 198阅读 0赞

在前面的文章中，我们对AQS（AbstractQueuedSynchronizer）进行了说明，AQS提供的是同步器的基本实现框架，在Java中我们并不是直接用AQS，因为它本身就是抽象的，我们只能用它的继承类。在这一篇文章中，我们就来看看几个基于AQS的同步器，仔细研究一下它们的原理及使用。

在JDK中，基于AQS实现的同步器主要有CountDownLatch、CyclicBarrier、Semaphore，下面我们一个个来看看它们的实现原理与用法。

（一）CountDownLatch

我们先从JDK中该类的说明看起，这才是最权威的解释。

![70][]

**作用和使用说明：**

在第一段，可以详细的看到这个类的作用说明：这个类提供一个同步化的机制，用于控制一个或多个线程等待，直到其他线程中的一组操作完成之后再继续执行。

单词Latch的中文翻译是门闩，也就是有“门锁”的功能，所以当门没有打开时，这N个人是不能进入屋内的，也就是这N个线程是不能继续往下运行的，将会被阻塞。

单词CountDown的中文翻译是倒计时，倒计时一定是从某个值开始往下递减，直到减到0才结束。CountDownLatch在初始化实例的时候需要传入一个count值，只有当这个count值降为0时，在这个同步器上阻塞的线程才能继续往下执行。

所以，CountDownLatch是通过一个计数器来实现的，计数器的初始化值为同步状态数量。每当一个线程完成了自己的任务后，就会消耗一个同步状态，计数器的值会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务了。

需要注意的是这里的同步状态数量是不可以重复使用的，是一次性的。

调用await方法将会根据同步状态数量count来决定是否阻塞，如果count变成了0，则阻塞的线程都会被唤醒并继续执行。调用countDown方法可以减小count的值。通过await和countDown方法的结合完成同步。

源码解读：

![70 1][]

从源码看，CountDownLatch提供了一个public的带int参数的构造方法，这里的参数就是同步状态数量。根据这个count构造了一个同步器Sync。

这里的Sync又是什么呢？跟随源码看看：

![70 2][]

从上面可以看出来，Sync是在CountDownLatch类内部定义的继承自AQS的静态同步器类。这个Sync类中主要实现了两个方法，tryAcquireShared(int acquires)和tryReleaseShared(int releases)方法。这里之所以实现这两个这两个方法，是因为这里实现的同步状态是非互斥的，可以允许多个线程共享，因此这里继承实现的是AQS的获取共享锁和释放共享锁的方法。

tryAcquireShared(int acquires)：这个方法查看当前的同步状态数量是否为0，如果是则返回1，表示获取成功，否则返回-1，表示获取失败。

tryReleaseShared(int releases)：通过自旋和CAS对同步状态进行减一操作，如果剩余的同步状态值为0，则标识同步状态已经释放，如果同步状态值>0，则标识同步状态还有线程在用，同步状态没有释放。

说完了Sync，我们回到正题CountDownLatch这个类，从上面知道，这个类是通过结合await和countDown方法结合使用的。先看下await方法。await方法有两种实现，一种是不带过期时间的，一种是带过期超时的。

![70 3][]

![70 4][]

![70 5][]

在await()方法中sync.acquireSharedInterruptibly(1)还是通过调用tryAcquireShared()方法来实现的，在countDown()方法中sync.releaseShared(1)是通过tryReleaseShared()实现的，如下所示：

![70 6][]

![70 7][]

至此，CountDownLatch的基本实现我们大概了解了。需要说明的是，AQS为我们实现了同步的整个框架，而各个继承类其实只需要根据情况实现相应的获取和释放逻辑即可，不需要太关注底层的同步状态的维护等一系列复杂逻辑。

（二）CyclicBarrier

前面的CountDownLatch，我们知道同步状态是一次性的，那么怎么才能做到复用同步状态呢？答案就是使用CyclicBarrier。

![70 8][]

根据上面类的解释说明，CyclicBarrier是一种同步化机制，用于实现一组线程相互等待并达到一个共同的障碍点。CyclicBarrier在编程实现需要固定线程数相互等待后同时进行某项操作的任务时非常有用。Barrier在所有等待线程释放后，是可以重复使用的，因此叫做Cyclic(循环）。

CyclicBarrier同时支持一个可选的Runnable参数，这个参数可以在所有需要到达障碍点的线程都到达之后执行一次，这个特性在这些线程共同达到障碍点后继续往后执行前更新共享状态非常有用。

![70 9][]

标号为1的代码，其中parties变量用于记录每次到达障碍点需要多少线程到达，count变量用于记录当前还剩下多少线程需要到达障碍点，每次当所有线程到达障碍点之后，count都会从parties重新赋值。标号为2的代码用于提供一个Runnable参数用于当所有线程到达障碍点后在积蓄往后执行前执行一个操作。

我们看一下await()方法，也是有两种（一种带超时时间），如下：

![70 10][]

![70 11][]

核心还是在于dowait()方法啊，我们来看看这段复杂的逻辑，在看着个核心代码前，先来了解下其他东西。为了记录当前代障碍是否已经被打破，在CyclicBarrier内部定义了一个内部类Generation，在Generation内部定义了一个broken变量，用于记录当前这次循环中障碍是否已经打破。每次需要重置的时候只要重新创建一个Generation对象即可，非常方便。

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)//如果当前Generation是处于打破状态则传播这个BrokenBarrierExcption
                    throw new BrokenBarrierException();
    
                if (Thread.interrupted()) {
                    breakBarrier();//如果当前线程被中断则使得当前generation处于打破状态，重置剩余count。并且唤醒状态变量。这时候其他线程会传播BrokenBarrierException.
                    throw new InterruptedException();
                }
    
               int index = --count;//尝试降低当前count
               if (index == 0) {  // tripped//如果当前状态将为0，则Generation处于开闸状态。运行可能存在的command，设置下一个Generation。相当于每次开闸之后都进行了一次reset。
                   boolean ranAction = false;
                   try {
                       final Runnable command = barrierCommand;
                       if (command != null)
                           command.run();
                       ranAction = true;
                       nextGeneration();
                       return 0;
                   } finally {
                       if (!ranAction)//如果运行command失败也会导致当前屏障被打破。
                           breakBarrier();
                   }
               }
    
                // 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();//这种捕获了InterruptException之后调用Thread.currentThread().interrupt()是一种通用的方式。但是之前源码中好像都没有体现。我第一次见这个好像是java并发实践中。这样做的目的是什么？其实就是为了保存中断状态，从而让其他更高层次的代码注意到这个中断。但是需要注意的是这里需要其他代码予以配合才行否则这样做其实是比较危险的一种方式，因为这相当于吞了这个异常。
                        }
                    }
    
                    //从阻塞恢复之后，需要重新判断当前的状态。
                    if (g.broken)
                        throw new BrokenBarrierException();
    
                    if (g != generation)
                        return index;
    
                    if (timed && nanos <= 0L) {
                        breakBarrier();
                        throw new TimeoutException();
                    }
                }
            } finally {
                lock.unlock();
            }
        }

private void breakBarrier() {
            generation.broken = true;
            count = parties;
            trip.signalAll();
        }

/**
         * Updates state on barrier trip and wakes up everyone.
         * Called only while holding lock.
         */
        private void nextGeneration() {
            // signal completion of last generation
            trip.signalAll();
            // set up next generation
            count = parties;
            generation = new Generation();
        }

如果要复用这个障碍器，那么需要通过reset方法可以重置障碍器。

/**
         * Resets the barrier to its initial state.  If any parties are
         * currently waiting at the barrier, they will return with a
         * {@link BrokenBarrierException}. Note that resets <em>after</em>
         * a breakage has occurred for other reasons can be complicated to
         * carry out; threads need to re-synchronize in some other way,
         * and choose one to perform the reset.  It may be preferable to
         * instead create a new barrier for subsequent use.
         */
        public void reset() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                breakBarrier();   // break the current generation
                nextGeneration(); // start a new generation
            } finally {
                lock.unlock();
            }
        }

（三）Semaphore

Semaphore，又名信号量。用于控制同时访问某个资源的线程数量。当超过阈值之后，后续访问的线程将会被阻塞，直到前面有线程释放掉自己的权限。

在Semaphore内部定义了三个内部类，分别为Sync、NonfairSync和FairSync。这三个内部类之间的关系如下：

![70 12][]

AbstratQueuedSynchronizer这个类我们在之前的文章中已经详细说过了，接下来，我们来看看着三个内部类。

首先是Sync，结合源码来看：

// 内部类，继承自AQS
        abstract static class Sync extends AbstractQueuedSynchronizer {
            // 版本号
            private static final long serialVersionUID = 1192457210091910933L;
            
            // 构造函数
            Sync(int permits) {
                // 设置状态数
                setState(permits);
            }
            
            // 获取许可
            final int getPermits() {
                return getState();
            }
    
            // 共享模式下非公平策略获取
            final int nonfairTryAcquireShared(int acquires) {
                for (;;) { // 无限循环
                    // 获取许可数
                    int available = getState();
                    // 剩余的许可
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining)) // 许可小于0或者比较并且设置状态成功
                        return remaining;
                }
            }
            
            // 共享模式下进行释放
            protected final boolean tryReleaseShared(int releases) {
                for (;;) { // 无限循环
                    // 获取许可
                    int current = getState();
                    // 可用的许可
                    int next = current + releases;
                    if (next < current) // overflow
                        throw new Error("Maximum permit count exceeded");
                    if (compareAndSetState(current, next)) // 比较并进行设置成功
                        return true;
                }
            }
    
            // 根据指定的缩减量减小可用许可的数目
            final void reducePermits(int reductions) {
                for (;;) { // 无限循环
                    // 获取许可
                    int current = getState();
                    // 可用的许可
                    int next = current - reductions;
                    if (next > current) // underflow
                        throw new Error("Permit count underflow");
                    if (compareAndSetState(current, next)) // 比较并进行设置成功
                        return;
                }
            }
    
            // 获取并返回立即可用的所有许可
            final int drainPermits() {
                for (;;) { // 无限循环
                    // 获取许可
                    int current = getState();
                    if (current == 0 || compareAndSetState(current, 0)) // 许可为0或者比较并设置成功
                        return current;
                }
            }
        }

Sync类中存在的方法和作用如下：

![70 13][]

紧接着是两个NonfairSync，NonfairSync继承自Sync，采用公平策略获取资源，类中只有一个方法tryAcquireShared(int acquires)，该方法重写了AQS中的同名方法，源码如下：

static final class NonfairSync extends Sync {
            // 版本号
            private static final long serialVersionUID = -2694183684443567898L;
            
            // 构造函数
            NonfairSync(int permits) {
                super(permits);
            }
            // 共享模式下获取
            protected int tryAcquireShared(int acquires) {
                return nonfairTryAcquireShared(acquires);
            }
        }

说明：从tryAcquireShared方法的源码可知，它会调用父类Sync的nonfairTryAcquireShared方法，表示按照非公平策略进行资源的获取。

FairSync类也继承了Sync类，表示采用公平策略获取资源，其只有一个tryAcquireShared方法，重写了AQS的该方法，其源码如下：

protected int tryAcquireShared(int acquires) {
                for (;;) { // 无限循环
                    if (hasQueuedPredecessors()) // 同步队列中存在其他节点
                        return -1;
                    // 获取许可
                    int available = getState();
                    // 剩余的许可
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining)) // 剩余的许可小于0或者比较设置成功
                        return remaining;
                }
            }

从tryAcquireShared方法的源码可知，它使用公平策略来获取资源，它会判断同步队列中是否存在其他的等待节点。

看完了这三个内部类，接下来继续看看Semaphore。

在构造Semaphore对象时，默认采用非公平策略获取资源，如下所示：

/**
         * Creates a {@code Semaphore} with the given number of
         * permits and nonfair fairness setting.
         *
         * @param permits the initial number of permits available.
         *        This value may be negative, in which case releases
         *        must occur before any acquires will be granted.
         */
        public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }

当然，也提供了构造参数可以控制想用什么样的策略。

/**
         * Creates a {@code Semaphore} with the given number of
         * permits and the given fairness setting.
         *
         * @param permits the initial number of permits available.
         *        This value may be negative, in which case releases
         *        must occur before any acquires will be granted.
         * @param fair {@code true} if this semaphore will guarantee
         *        first-in first-out granting of permits under contention,
         *        else {@code false}
         */
        public Semaphore(int permits, boolean fair) {
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

接下来看下Semaphore获取同步状态的入口acquire(int permits)，通过调用AQS的acquireSharedInterruptibly方法获取同步状态，这个方法又转为调用NonfairSync或FairSync的tryAcquireShared方法。两者的区别在于对于公平模式下获取资源时会首先判断同步队列中是否有其他节点，因为公平模式下需要遵循FIFO，前面如果有节点等待获取同步状态，那么这时候后面不能让其优先获取同步状态，如果是非公平模式，那么不会关注同步队列中是否有其他节点，而是只关注剩余的同步状态量。如下：

![70 14][]

对于同步状态的释放，Semaphore的release()方法进行释放。如下所示：

/**
         * Releases a permit, returning it to the semaphore.
         *
         * <p>Releases a permit, increasing the number of available permits by
         * one.  If any threads are trying to acquire a permit, then one is
         * selected and given the permit that was just released.  That thread
         * is (re)enabled for thread scheduling purposes.
         *
         * <p>There is no requirement that a thread that releases a permit must
         * have acquired that permit by calling {@link #acquire}.
         * Correct usage of a semaphore is established by programming convention
         * in the application.
         */
        public void release() {
            sync.releaseShared(1);
        }

这里，底层调用了AQS的releaseShared()方法，如下所示，转换为了tryReleaseShared()方法和doReleaseShared()方法，而tryReleaseShared在AQS的继承类Sync中进行了重新实现。

/**
         * Releases in shared mode.  Implemented by unblocking one or more
         * threads if {@link #tryReleaseShared} returns true.
         *
         * @param arg the release argument.  This value is conveyed to
         *        {@link #tryReleaseShared} but is otherwise uninterpreted
         *        and can represent anything you like.
         * @return the value returned from {@link #tryReleaseShared}
         */
        public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {
                doReleaseShared();
                return true;
            }
            return false;
        }

tryReleaseShared（）方法如下：

protected final boolean tryReleaseShared(int releases) {
                for (;;) {
                    int current = getState();
                    int next = current + releases;
                    if (next < current) // overflow
                        throw new Error("Maximum permit count exceeded");
                    if (compareAndSetState(current, next))
                        return true;
                }
            }

如果释放成功了，还要通过调用AQS的doReleaseShared方法来完成对同步等待队列中节点同步状态的更新，源码如下：

private void doReleaseShared() {
            /*
             * Ensure that a release propagates, even if there are other
             * in-progress acquires/releases.  This proceeds in the usual
             * way of trying to unparkSuccessor of head if it needs
             * signal. But if it does not, status is set to PROPAGATE to
             * ensure that upon release, propagation continues.
             * Additionally, we must loop in case a new node is added
             * while we are doing this. Also, unlike other uses of
             * unparkSuccessor, we need to know if CAS to reset status
             * fails, if so rechecking.
             */
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }

至此，基于AQS实现的常用并发类便简单介绍完成了。

当然并发包中有很多其他类，需要了解的可以自己看看源码，对于并发底层的实现个人也还是有点不太清楚，还需要花时间来继续研究，受个人能力有限，只能说明如此了哈哈。

感谢大家的阅读，如果有对Java编程、中间件、数据库、及各种开源框架感兴趣，欢迎关注我的博客和头条号（源码帝国），博客和头条号后期将定期提供一些相关技术文章供大家一起讨论学习，谢谢。

如果觉得文章对您有帮助，欢迎给我打赏，一毛不嫌少，一百不嫌多，^\_^谢谢。

![20180312232547620][]

[70]: /images/20220528/c72a191df9124c05b1209ef5b9524e66.png
[70 1]: /images/20220528/5890ce0028534932acbb78373ab95735.png
[70 2]: /images/20220528/96cc7c4d4ae04474b7fe582768ce9893.png
[70 3]: /images/20220528/35f8979d033e4648bd90f9efedda5426.png
[70 4]: /images/20220528/d24ca4f3d9bc4b448c9d684ed0a170a0.png
[70 5]: /images/20220528/787d5ef469274db796a020811aa75d14.png
[70 6]: /images/20220528/8c0f6b7f04e74f17824cd175672633e0.png
[70 7]: /images/20220528/6ac89d267c2a411d8522e1f46dd310c6.png
[70 8]: /images/20220528/a0492a4cb5c042e5a14c1754d9935baf.png
[70 9]: /images/20220528/f9e8320907204fd393193af46d697fe4.png
[70 10]: /images/20220528/542fc7a70aaf48fda1be9890efe4f2f6.png
[70 11]: /images/20220528/daae4fdf84bd40148601b42d464c3c77.png
[70 12]: /images/20220528/de7cc136b8ac48259253638c6a79b74c.png
[70 13]: /images/20220528/11529b0ebc2048869882bafcec1b0edb.png
[70 14]: /images/20220528/1d5a019fa8744b86a64639c0c0919c11.png
[20180312232547620]: /images/20220528/9b10dca4afe642bfba1a534466a895d0.png