上一篇主要结束AQS的独占功能，这次我们通过对CountDownLathc的介绍来解读AQS的另一个功能：共享功能。

AQS共享功能的实现

我们将用CountDownLatch来阐述AQS共享功能，关于CountDownLatch的使用，可以参见以前写的博客。在多线程环境下，需要等待的线程调用CountDownLatch的await()方法会被阻塞，等到其他线程调用countDown()方法后，将计数器减为0时，被阻塞的线程才会唤醒。具体是一个什么样的过程，可以用下面这个简单程序debug一下，然后可以参考这篇多线程程序调试指南：

package com.test;

import java.util.concurrent.CountDownLatch;

public class Test {

    public static void main(String[] args) {
        final CountDownLatch countDownLatch = new CountDownLatch(2);
        
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                countDownLatch.countDown();
            }
        }).start();
        
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                countDownLatch.countDown();
            }
        }).start();
        
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

关于CountDownLatch的实现，我们首先可以看它的构造方法：

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

和ReentrantLock类似，CountDownLatch内部也有一个Sync的内部类，同样也是继承了AQS。

再来看一下Sync：

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

上一篇讲到setState()方法是AQS的一个“状态位”，在ReentrantLock中，表示加锁的次数，而在CountDownLatch中，表示计数器的初始大小。我们点击进入setState()方法，进入到AQS：

    /**
     * The synchronization state.
     */
    //这个state是volatile变量，在多个线程中共享
    private volatile int state;

    /**
     * Returns the current value of synchronization state.
     * This operation has memory semantics of a {@code volatile} read.
     * @return current state value
     */
    protected final int getState() {
        return state;
    }

    /**
     * Sets the value of synchronization state.
     * This operation has memory semantics of a {@code volatile} write.
     * @param newState the new state value
     */
    protected final void setState(int newState) {
        state = newState;
    }

await()

设置万计数器大小后，CountDownLatch的构造方法返回，我们接着看CountDownLatch 的await()方法：

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

调用了Sync的acquireSharedInterruptibly()方法，实际上是调用AQS的acquireSharedInterruptibly()方法：

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

从方法名上看，这个方法的调用是响应线程的打断的，所以在前两行会检查下线程是否被打断。接着，尝试着获取共享锁，小于 0，表示获取失败，通过本系列的上半部分的解读， 我们知道 AQS 在获取锁的思路是，先尝试直接获取锁，如果失败会将当前线程放在队列中，按照 FIFO 的原则等待锁。而对于共享锁也是这个思路，如果和独占锁一致，这里的 tryAcquireShared 应该是个空方法，留给子类去判断:

    protected int tryAcquireShared(int arg) {
        throw new UnsupportedOperationException();
    }

接着看CountDownLatch的tryAcquireShared()：

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

如果state变成0，则返回1，表示获取成功，否则返回-1，表示获取失败。

看到这里，我们可能会异或，await()方法的获取方法更像在获取独占锁，为什么还用tryAcquireShared()呢？

回想一下CountDownLatch的await()方法是不是智能在主线程中调用？并不是，CountDownLatch的await()方法可以在多个线程中调用，当CountDownLatch的计数器为0后，调用 await 的方法都会依次返回。 也就是说可以多个线程同时在等待 await 方法返回，所以它被设计成了实现 tryAcquireShared 方法，获取的是一个共享锁，锁在所有调用 await 方法的线程间共享，所以叫共享锁。

回到 acquireSharedInterruptibly 方法：

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

如果获取共享锁失败（返回了-1，说明state不为0，即CountDownLatch的计数器还不为0），进入调用 doAcquireSharedInterruptibly() 方法，将当前线程放入到队列中。

AQS的结构，上一篇已经详细阐述了，本质是一个双向链表，可以参考上一篇文章。

继续进入到doAcquireSharedInterruptibly()方法：

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED); 
// 将当前线程包装为类型为 Node.SHARED 的节点，标示这是一个共享节点。
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
// 如果新建节点的前一个节点，就是 Head，说明当前节点是 AQS 队列中等待获取锁的第一个节点，
// 按照 FIFO 的原则，可以直接尝试获取锁。
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
// 获取成功，需要将当前节点设置为 AQS 队列中的第一个节点，这是 AQS 的规则 // 队列的头节点表示正在获取锁的节点 
                        setHeadAndPropagate(node, r); 
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) && // 检查下是否需要将当前节点挂起 
                    parkAndCheckInterrupt()) 
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

这里的代码说明：
1.setHeadAndPropagete()方法：

    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        /*
         * Try to signal next queued node if:
         *   Propagation was indicated by caller,
         *     or was recorded (as h.waitStatus either before
         *     or after setHead) by a previous operation
         *     (note: this uses sign-check of waitStatus because
         *      PROPAGATE status may transition to SIGNAL.)
         * and
         *   The next node is waiting in shared mode,
         *     or we don't know, because it appears null
         *
         * The conservatism in both of these checks may cause
         * unnecessary wake-ups, but only when there are multiple
         * racing acquires/releases, so most need signals now or soon
         * anyway.
         */
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

首先，使用CAS更好头结点，然后将当前结点的下一个结点取出来，如果同样是“shared”类型的，在做一次“releaseShared”操作。

再看一下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;
              // 如果当前节点是 SIGNAL 意味着，它正在等待一个信号，  
              // 或者说，它在等待被唤醒，因此做两件事，1 是重置 waitStatus 标志位，2 是重置成功后, 唤醒下一个节点。
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                // 如果本身头节点的 waitStatus 是出于重置状态（waitStatus==0）的，    将其设置为“传播”状态。
              // 意味着需要将状态向后一个节点传播。
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

这样做的原因是：共享功能和独占功能不一样的地方，对于独占功能，有且只有一个线程（通常只对应一个结点，拿ReentrantLock来说，如果当前持有锁的线程重复调用lock()方法，会被包装成多个节点在 AQS 的队列中，所以用一个线程来描述更准确），能够获取锁。

但是对于共享功能来说，共享的状态是可以共享的，也就是意味着其他AQS队列中的其他结点也能第一时间知道状态的变化。因此，一个节点获取到共享状态流程图是这样的：

比如现在有如下队列：
当 Node1 调用 tryAcquireShared 成功后，更换了头节点：

Node1 变成了头节点然后调用 unparkSuccessor() 方法唤醒了 Node2、Node2 中持有的线程 A 出于上面流程图的 park node 的位置，线程 A 被唤醒后，重复黄色线条的流程，重新检查调用 tryAcquireShared 方法，看能否成功，如果成功，则又更改头节点，重复以上步骤，以实现节点自身获取共享锁成功后，唤醒下一个共享类型节点的操作，实现共享状态的向后传递。

2.对于doAcquireShared()方法，AQS还提供了类似的实现：

分别对应了：
1）带参数请求共享锁。 （忽略中断）
2）带参数请求共享锁，且响应中断。（每次循环时，会检查当前线程的中断状态，以实现对线程中断的响应）
3）带参数请求共享锁但是限制等待时间。（第二个参数设置超时时间，超出时间后，方法返回。）

countDown()

看完await()方法，接着看countDown()方法：

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

调用了AQS的releaseShared方法，并传入了参数1：

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

先尝试释放锁，tryReleaseShared 同样为空方法，留给子类自己去实现，以下是 CountDownLatch 的内部类 Sync 的实现：

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

死循环更新 state 的值，实现 state 的减 1 操作，之所以用死循环是为了确保 state 值的更新成功。

从上文的分析中可知，如果 state 的值为 0，在 CountDownLatch 中意味：所有的子线程已经执行完毕，这个时候可以唤醒调用 await() 方法的线程了，而这些线程正在 AQS 的队列中，并被挂起的，

所以下一步应该去唤醒 AQS 队列中的头节点了（AQS 的队列为 FIFO 队列），然后由头节点去依次唤醒 AQS 队列中的其他共享节点。

如果 tryReleaseShared 返回 true, 进入 doReleaseShared() 方法：

private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) { 
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 
// 如果当前节点是 SIGNAL 意味着，它正在等待一个信号，
 // 或者说，它在等待被唤醒，因此做两件事，1 是重置 waitStatus 标志位，2 是重置成功后, 唤醒下一个节点。
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))  
// 如果本身头节点的 waitStatus 是出于重置状态（waitStatus==0）的，将其设置为“传播”状态。
// 意味着需要将状态向后一个节点传播。
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
  }

当线程被唤醒后，会重新尝试获取共享锁，而对于 CountDownLatch 线程获取共享锁判断依据是 state 是否为 0，而这个时候显然 state 已经变成了 0，因此可以顺利获取共享锁并且依次唤醒 AQS 队里中后面的节点及对应的线程。

总结

本文从 CountDownLatch 入手，深入分析了 AQS 关于共享锁方面的实现方式：

如果获取共享锁失败后，将请求共享锁的线程封装成 Node 对象放入 AQS 的队列中，并挂起 Node 对象对应的线程，实现请求锁线程的等待操作。待共享锁可以被获取后，从头节点开始，依次唤醒头节点及其以后的所有共享类型的节点。实现共享状态的传播。

这里有几点值得注意：

1.与 AQS 的独占功能一样，共享锁是否可以被获取的判断为空方法，交由子类去实现。
2.与 AQS 的独占功能不同，当锁被头节点获取后，独占功能是只有头节点获取锁，其余节点的线程继续沉睡，等待锁被释放后，才会唤醒下一个节点的线程，而共享功能是只要头节点获取锁成功，就在唤醒自身节点对应的线程的同时，继续唤醒 AQS 队列中的下一个节点的线程，每个节点在唤醒自身的同时还会唤醒下一个节点对应的线程，以实现共享状态的“向后传播”，从而实现共享功能。
以上的分析都是从 AQS 子类的角度去看待 AQS 的部分功能的，而如果直接看待 AQS，或许可以这么去解读：

首先，AQS 并不关心“是什么锁”，对于 AQS 来说它只是实现了一系列的用于判断“资源”是否可以访问的 API, 并且封装了在“访问资源”受限时将请求访问的线程的加入队列、挂起、唤醒等操作， AQS 只关心“资源不可以访问时，怎么处理？”、“资源是可以被同时访问，还是在同一时间只能被一个线程访问？”、“如果有线程等不及资源了，怎么从 AQS 的队列中退出？”等一系列围绕资源访问的问题，而至于“资源是否可以被访问？”这个问题则交给 AQS 的子类去实现。

当 AQS 的子类是实现独占功能时，例如 ReentrantLock，“资源是否可以被访问”被定义为只要 AQS 的 state 变量不为 0，并且持有锁的线程不是当前线程，则代表资源不能访问。

当 AQS 的子类是实现共享功能时，例如：CountDownLatch，“资源是否可以被访问”被定义为只要 AQS 的 state 变量不为 0，说明资源不能访问。

这是典型的将规则和操作分开的设计思路：规则子类定义，操作逻辑因为具有公用性，放在父类中去封装。

当然，正式因为 AQS 只是关心“资源在什么条件下可被访问”，所以子类还可以同时使用 AQS 的共享功能和独占功能的 API 以实现更为复杂的功能。

比如：ReentrantReadWriteLock，我们知道 ReentrantReadWriteLock 的中也有一个叫 Sync 的内部类继承了 AQS，而 AQS 的队列可以同时存放共享锁和独占锁，对于 ReentrantReadWriteLock 来说分别代表读锁和写锁，当队列中的头节点为读锁时，代表读操作可以执行，而写操作不能执行，因此请求写操作的线程会被挂起，当读操作依次推出后，写锁成为头节点，请求写操作的线程被唤醒，可以执行写操作，而此时的读请求将被封装成 Node 放入 AQS 的队列中。如此往复，实现读写锁的读写交替进行。

而本系列文章上半部分提到的 FutureTask，其实思路也是：封装一个存放线程执行结果的变量 A, 使用 AQS 的独占 API 实现线程对变量 A 的独占访问，判断规则是，线程没有执行完毕：call() 方法没有返回前，不能访问变量 A，或者是超时时间没到前不能访问变量 A(这就是 FutureTask 的 get 方法可以实现获取线程执行结果时，设置超时时间的原因)。

参考资料

https://www.infoq.cn/article/java8-abstractqueuedsynchronizer