CountDownLatch源码分析

使用方法

CountDownlatch在初始化的时候，需要赋予一个初始值，用来初始化计数器，然后在子线程里面调用countDown方法来进行减一，当计数器的值为0的时候，调用countdownlatch.await()的线程才能够继续往下走，在值为0之前，调用await的线程会一直被阻塞。

下面的实例代码中，由于在初始化CountDownlatch的时候传入的值是100，但是实际只被减了99次，所以主线程会一直被阻塞住

public class TestCountDownLatch
{
    private static CountDownLatch countDownLatch = new CountDownLatch(100);
    
    private static void costCountdownLatch()
    {
        System.out.println(Thread.currentThread().getName());
        countDownLatch.countDown();
    }
    
    public static void main(String[] args) throws Exception
    {
        System.out.println("开始跑");
        for (int i = 0; i < 99; i++)
        {
            new Thread(TestCountDownLatch::costCountdownLatch).start();
        }
        System.out.println("开始等待");
        countDownLatch.await();
        System.out.println("结束");
    }
    
}

源码

CountDownLatch里面使用AQS来做线程之间同步的

Sync

使用传入的count来作为AQS中的state进行同步状态的维护

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) {
            return (getState() == 0) ? 1 : -1;
        }

        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;
            }
        }
    }

初始化函数

根据传入的值，设置AQS的state变量

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

await

AQS源码分析 await调用了AQS的共享获取，并且使用了上面重写的tryAcquireShared。

能够从上面的代码看到，在state不等于0的时候，一直都是返回-1，也就是获取锁失败

后面的acquire就是AQS里面的acquire方法

public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
        if (Thread.interrupted() ||
            (tryAcquireShared(arg) < 0 &&
             acquire(null, arg, true, true, false, 0L) < 0))
            throw new InterruptedException();
    }

countDown

对计数器减一，如果计数器已经等于0了，就通知等待的线程

public void countDown() {
        sync.releaseShared(1);
    }

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            signalNext(head);
            return true;
        }
        return false;
    }