JDK类库源码分析系列2-AbstractQueuedSynchronizer-CountDownLatch

Posted 2021-05-24 _微风轻起

一、简介说明

​ 这一篇我们来介绍下AbstractQueuedSynchronizer的一个应用类CountDownLatch。其是用来进行计数的，调用其的countDownLatch.await()方法后，其一定会等到其初始化的计算为0才往下继续运行。

​ 例如我们一个demo：

public static void main(String[] args) {
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Runnable 1 -------------");
        }
    }).start();
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Runnable 2 -------------");
        }
    }).start();
    System.out.println("Main Runnable -----------");
}

​ 这个方法正常运行结果是下面这个：

Main Runnable -----------
Runnable 1 -------------
Runnable 2 -------------

​ 我们简单改造下来说明这个CountDownLathc的用法：

public static void main(String[] args) throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(2);
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Runnable 1 -------------");
            countDownLatch.countDown();
        }
    }).start();
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Runnable 2 -------------");
            countDownLatch.countDown();
        }
    }).start();
    countDownLatch.await();
    System.out.println("Main Runnable -----------");
}

​ 其的运行结果就会如下：

Runnable 2 -------------
Runnable 1 -------------
Main Runnable -----------

​ 这里的代码我们要注意这里用的是countDownLatch.await()，await方法，而不是Object的wait方法。

​ 下面我们就来具体梳理下这个类。

一、基本结构

1、CountDownLatch

public class CountDownLatch {

​ 可以看到这个CountDownLatch的继承实现关系很简单，因为其并没有继承&实现其他的类或接口。

二、变量

1、sync

private final Sync sync;

​ 其变量就一个sync。

三、Sync子类

1、结构

private static final class Sync extends AbstractQueuedSynchronizer {

​ 我们可以看到其是直接继承的AQS。

2、方法

1）、Sync(int count)

Sync(int count) {
    setState(count);
}

​ 这个构造方法是直接将入参赋值给AQS的status属性。

2）、getCount()

int getCount() {
    return getState();
}

​ 获取status的值。

3）、tryAcquireShared(int acquires)

protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

​ 用共享的模式尝试获取锁。

4）、tryReleaseShared(int releases)

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

​ 以共享模式释放锁。其是就是将status的值减一，我们可以看到这里其是入参releases是没有用的，表示一次只能减一，同时这里是使用了for(;;)自旋以及CAS来修改status。

四、方法

1、CountDownLatch(int count)

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

​ 构造方法，可以看到其就是初始化sync，使用入参count来设置status。

2、await()

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

​ 阻塞订单方法，可以看到其是能抛出InterruptedException异常的。

​ 然后其是直接调用CAS的acquireSharedInterruptibly方法。

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    }
}

​ 我们知道在Lock锁结构中，在上锁过程中我们的动作会包含tryAcquire()尝试获取，所以这里其实我们可以将CountDownLatch的await()等待看做是上锁过程。然后下面的countDown()看做是释放锁的方法。

3、await(long timeout, TimeUnit unit)

public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

​ 等待固定长的时间

4、countDown()

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

​ 释放资源，也就是将status的值减一。这里调用的CouuntDownLatch类实现的releaseShared()方法。

5、getCount()

public long getCount() {
    return sync.getCount();
}

​ 获取status的值，也可以理解为还有多少资源没有释放。