CountDownLatch & Semaphore 实现原理详解

「爱情、让人受尽委屈。」 2022-05-28 01:51 187阅读 0赞

CountDownLatch 和 Semaphore 是Java中常用的两个同步器。他们的用法百度一下一大堆，我就不多做介绍了。下面我将从源码的角度分析一下这两个类的实现原理。

阅读本篇文章之前，建议先理解AQS同步器的原理。可以看我之前写的一篇文章:  
[Java同步器框架-AQS原理&源码解析][Java_-AQS]

理解了AQS的底层原理，再来看CountDownLatch 和 Semaphore 的实现原理就会发现很简单。

## CountDownLatch 实现原理 ##

CountDownLatch 内部定义了一个AQS的实现类

private final Sync sync;
    private static final class Sync extends AbstractQueuedSynchronizer { 
            private static final long serialVersionUID = 4982264981922014374L;
    
            Sync(int count) {
                setState(count);
            }
    
            int getCount() {
                return getState();
            }
    
            protected int tryAcquireShared(int acquires) {
                //判断state是否为0了，如果为0就返回1，表示允许通过。返回-1的话线程就要挂起等待state为0
                return (getState() == 0) ? 1 : -1;
            }
    
            protected boolean tryReleaseShared(int releases) {
                //通过自旋+CAS的操作将state-1。当state减到0的时候AQS就会去唤醒挂起中的线程
                for (;;) {
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
    }
    public CountDownLatch(int count) {
            if (count < 0) throw new IllegalArgumentException("count < 0");
            //初始化AQS同步器
            this.sync = new Sync(count);
    }
    public void await() throws InterruptedException {
            //尝试获取资源,等待的时候支持线程响应中断
            sync.acquireSharedInterruptibly(1);
    }
    public void countDown() {
            //释放资源
            sync.releaseShared(1);
    }

可以看出,countDown()和await()方法的底层其实都是依赖于Sync。

Sync是一个AQS同步器，实现了 tryAcquireShared()和tryReleaseShared()，这是一个**共享模式**下的同步器。

CountDownLatch在初始化的时候传入一个同步线程数量count表示AQS的state。然后执行await()的时候，AQS会先判断state是否为0，state不为0的话就会阻塞当前线程，并让当前线程进入AQS的CLH队列中排队。

当有线程执行countDown的时候,Sync会通过cas+自旋的方式将state减一，然后判断state是否等于0。等于0的时候返回true，AQS发现tryReleaseShared()返回true，就会去唤醒正在CLH队列中排队等待的线程，先唤醒排在最前面的那个线程。由于是共享模式，那个线程被唤醒后，检查state=0了，就结束阻塞，并且会通知下一个排队线程，下一个线程醒来后，一样判断state是否等于0了，然后结束阻塞，通知下一个，一直循环下去，直到所有阻塞中的线程全部被唤醒。

## Semaphore 实现原理 ##

也是基于AQS实现的。内部也定义了一个Sync ，AQS的实现类

private final Sync sync;
    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) {
                //用CAS+自旋的方式获取资源
                for (;;) {
                    //直接去获取资源
                    //假设现在CLH队列中有人在排队了，然后刚好有线程释放了一个资源并且唤醒队列中等待的线程了
                    //然后唤醒后还没来得及获取到资源，就被人插队获取了资源
                    //下面的代码就是尝试不排队直接获取资源，获取不到资源再去排队，所以这是一种不公平的算法
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }
    
            //释放资源
            protected final boolean tryReleaseShared(int releases) {
                //用CAS+自旋的方式释放资源
                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))
                        return current;
                }
            }
        }
    
        /** * NonFair version */
        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);
            }
        }
    
        /** * Fair version */
        static final class FairSync extends Sync {
            private static final long serialVersionUID = 2014338818796000944L;
    
            FairSync(int permits) {
                super(permits);
            }
    
            protected int tryAcquireShared(int acquires) {
                //用CAS+自旋的方式尝试获取资源
                for (;;) {
                    //判断CLH队列中是否有人在排队等待,如果是的话直接返回-1，表示也进入队列等待
                    if (hasQueuedPredecessors())
                        return -1;
                    //如果队列中没有人在排队，就尝试获取资源了
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }
    }
    //默认选择非公平的AQS同步器
        public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }
          //可以选择公平的还是不公平的同步器
        public Semaphore(int permits, boolean fair) {
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }
    
    public void acquire() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
    }
    
    public void release() {
            sync.releaseShared(1);
    }

在Semaphore 的例子中，AQS的state就可以直接看成是一种**资源(许可证)**。`acquire()`方法就是尝试获取资源，当资源不足的的时候，就进入CLH队列排队等待,也就是阻塞当前线程。

`release()`方法就是释放一些资源，释放完后就会通知CLH队列中的线程。CLH队列中的线程被唤醒后就会再次检查是否有资源了，有的话就消耗资源，也就是state-n,然后就退出阻塞。由于是**共享模式的AQS**，线程被唤醒后还会继续通知下一个排在自己后面的线程，那个线程也醒来检查资源是否足够，以此类推下去。

比较有意思的是，Semaphore 内部提供了两个AQS同步器的实现类，一种是公平模式的，一种是非公平模式的。这两种同步器其实只是在获取资源的行为上有所不同，也就是`tryAcquireShared(int acquires)`的实现不同，而在释放资源的行为上是一样的。

在公平模式下，获取资源前会先去检查CLH队列是否有人在排队了，如果发现有人在排队了，就会自觉也去排队。而在非公平模式下，它会直接尝试先获取资源，获取不到再去排队。

我们可以理解为一间店在卖面包，面包的数量会不断产生，如果数量不够了，排队的人就要在那等着。在公平模式下，有一个人来买面包，发现大家都在排队，就自觉进入队列中也排队等待。在非公平的模式下，来买面包的人就直接先去店主那看有没有剩余的面包可以买，这时候可以店主刚生产出一个面包来，还没来得及交给排在队伍的第一个人，就被插队者买走了。所以，这就是不公平的模式，违反了先到先得的规则。

## 总结 ##

CountDownLatch 和 Semaphore这两个类的原理其实都很好理解。看这两个类的源码也很快，结合AQS的原理一起理解，会对Java线程之间的同步有更深的理解。

最后，如果哪里有写的不对的地方，烦请指出，感激不尽！

[Java_-AQS]: https://blog.csdn.net/u013332124/article/details/79701415