Java源码学习之高并发编程基础——AQS源码剖析之阻塞队列（下）

1.前言&目录

前言：

在上一篇文章AQS源码剖析之阻塞队列（上）中介绍了以独占锁模式下AQS的基本原理，AQS仅仅起到了一个“维持线程等待秩序”的作用，那么本篇文章继续讲解共享锁模式下的特点。

AQS不操纵锁的获取或者释放，仅仅提供一个由双向链表组成的队列，让抢不到锁的线程进入队列排队并阻塞起来、持有锁的线程释放锁后“通知”（即从阻塞态中唤醒）排名最靠前的有效（非CANCELLED状态）节点去重新竞争锁资源。

使用独占锁锁住代码块是保证数据正确性的一种做法，锁住的代码块每次只能有一个线程访问，获取不到锁的线程会被阻塞住，极大的保证了数据正确性。

本文将讲解AQS独占锁的另一个“反例”——共享锁，还是以源码解读等形式进行一个深入理解和学习，理解它的应用场景以及能解决什么问题。

1.前言&目录

2.AQS独占锁回顾

2.1 入队

2.2 出队

3. AQS共享锁源码剖析

3.1 共享锁入队

3.1.1 acquireShared方法源码剖析

3.1.2 acquireSharedInterruptibly方法源码剖析

3.1.3 tryAcquireSharedNanos方法源码剖析

3.1.4 共享锁入队总结

3.2 共享锁出队

4.案例

5.总结

2.AQS独占锁回顾

回顾AQS对于多线程抢夺独占锁对应的入队和出队场景，细节如下：

2.1 入队

AQS独占锁的入队，背景是多个线程在并发抢锁时，由于锁只有一把，因此拿不到锁的线程会被封装为一个等锁节点并加入到队列。

AQS#acquire(int arg)方法是独占锁的获取、加入队列的实现，让我们回顾其源码，它里面调用了三个非常重要的方法：

    public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}

①tryAcquire(int arg)方法，它是由子类实现的钩子方法，用作判断当前线程能否获取独占锁，传入的arg参数被应用于AQS的state变量中，（state）其实是一个锁持有计数器，分为0和非0两种情况，0表示独占锁还没有被获取，非0表示锁已经被持有中。该方法返回true表示当前线程拿到了锁，false表示拿不到。
②addWaiter(Node mode)方法，它是AQS的实现，目的是为了将经过①后拿不到锁的线程封装为一个等待独占锁的节点并添加到队列尾部。这个过程由于存在并发操作，因此添加到尾部是通过自旋&CAS操作的。
③acquireQueued(final Node node, int arg)方法，这个方法入参是arg和②创建的等待节点，方法内部会进行自旋操作：根据arg参数二次调用tryAcquire(arg)方法再尝试获取锁。但是自旋次数是有限制的，如果二次拿锁失败，则会进行线程阻塞预判操作，下一次自旋中仍然拿不到锁，那么经过第二次的预判操作后就会将当前线程阻塞起来。而唤醒的时机是当其他持有锁的线程释放锁了，找到队列排名最靠前的（并且有效的）等待节点将它唤醒，重新进入到这里的自旋操作去获取锁。

2.2 出队

AQS独占锁的出队，就是排名最靠前的等待节点拿到锁后，移除旧头节点并将该等待节点升级为新头节点的过程。

出队的情况其实有两种，第一种就是在入队时调用acquireQueued方法进行自旋的过程中二次调用获取了锁，但是这种情况概率比较小，不常见。

最常见的是第二种，持有锁的线程主动释放锁，如下伪代码所示：

    public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}

独占锁的释放是AQS#release(int arg)完成的，它的执行框架是先调用子类的tryRelease(int arg)方法，如果返回true表示当前的独占锁从持有线程解除了，接着以当前的头节点为起点，继续去调用unparkSuccessor(Node node)，在这里面会找到头节点后面的有效后继节点（只会找一个），将阻塞中的线程唤醒重新加入acquireQueued方法的自旋中去。

由于这时候是持有线程主动释放独占锁，因此重新自旋时会极大可能拿到独占锁，然后移除旧头节点、升级为新头节点，出队完成。

3. AQS共享锁源码剖析

从字面意思上，共享锁和独占锁是对立的关系，独占锁是任一时刻只能由一个线程持有，实际上共享锁应该是独占锁的一种补充：共享锁在任一时刻可由多个线程持有，这便是”共享“。

的确从共享锁的实现类Semaphore、ReentrantReadWriteLock、CountDownLatch、LimitLatch来看，它们对于共享锁的获取的确不仅仅限于单个线程。

本章节，仍然会以共享锁的入队和出队去进行源码解读，不过共享锁的应用场景比独占锁的要多，因此让我们看看当了解其核心原理后，这些共享锁能应用到什么场景？

3.1 共享锁入队

共享锁的入队和独占锁的入队主要流程相差无异，均是在各自的锁被抢占完了以后，后面的线程拿不到而需要进行进入队列等待。

AQS提供了三种获取共享锁的方法：acquireShared(int arg)、acquireSharedInterruptibly(int arg)、tryAcquireSharedNanos(int arg, long nanosTimeout)，并且它们里面的调用链路都是一样的，可以抽象为如下伪代码：

if(tryAcquireShared(arg)<0){ // 返回结果小于0表示拿不到共享锁doAcquiredShared...(arg); // 尝试获取共享锁
}

因此，可以总结tryAcquireShared(arg)返回结果小于0时获取不到共享锁，我们需要记住这一结论。

那么接下来依次分析这三个获取共享锁的方法。

3.1.1 acquireShared方法源码剖析

acquireShared方法中是委托doAcquireShared(arg)去完成共享锁的获取。

在doAcquireShared方法也是分三步走：

①创建一个共享锁等待节点添加到队列尾部。
②进行自旋操作：二次获取锁，拿到锁以后将当前等待节点升级为新头节点，抛弃旧头节点，即其中一种出队方法；并且相比于独占锁模式，多了一个传播行为——setHeadAndPropagate。
③最多自旋两次仍然拿不到共享锁的话，则将当前线程阻塞等待唤醒。

        public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0)doAcquireShared(arg);}private void doAcquireShared(int arg) {final Node node = addWaiter(Node.SHARED);//创建一个共享锁等待节点添加到队列尾部boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head) {// 二次获取锁int r = tryAcquireShared(arg);if (r >= 0) {// 设置新头节点并执行传播行为setHeadAndPropagate(node, r);p.next = null; // help GCif (interrupted)selfInterrupt();failed = false;return;}}// 线程预判阻塞操作if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}

这“三部曲”和独占锁模式下几乎一模一样，唯一不同的是，在等待节点拿到锁升级为头节点后，还会执行一个传播行为，那么这个传播行为指的是什么呢？

其实传播行为指的是，由于共享锁能被多个线程同时持有，那么也会存在并发释放共享锁的情况，那么此时通知后面（阻塞中）的等待节点、唤醒它们就是一个更好的选择。独占锁不需要传播行为的原因是它的整个过程是这样的：拿锁-释放锁-拿锁-释放锁-拿锁-释放锁，都是紧紧按着顺序来的。而共享锁则是这样的：拿锁-释放锁-释放锁-拿锁-释放锁等，因此加入传播行为能更快的通知等待节点并唤醒它们重新自旋拿锁。

public abstract class AbstractQueuedSynchronizerextends AbstractOwnableSynchronizerprivate void setHeadAndPropagate(Node node, int propagate) {Node h = head; // Record old head for check belowsetHead(node);// propagate是判断当前共享锁是否空闲的依据，大于0肯定是共享锁“空闲”了if (propagate > 0 || h == null || h.waitStatus < 0 ||(h = head) == null || h.waitStatus < 0) {Node s = node.next;if (s == null || s.isShared())doReleaseShared();}}
}

上述伪代码就是设置新请求头和传播新为的执行，但是执行的条件似乎有点复杂，但是请注意这里的propagate变量是上层调用方doAcquireShared方法中调用的tryAcquredShared方法的返回结果，即大于0时表明此时共享锁”空闲“了，可以让其他线程来抢锁；其次还有当此时头节点的waitStatus小于0时（SIGNAL或者PROPAGATE），会找到当前（已经拿到共享锁的）等锁节点的后继节点，如果该后继节点是null或者共享节点则进一步调用doReleaseShared()方法——该方法是共享锁的出队底层实现，本小节不会讲解。即如果进入到if代码段的doReleaseShared()方法就完成了传播行为。

3.1.2 acquireSharedInterruptibly方法源码剖析

acquireSharedInterruptibly方法和acquireShared的逻辑相似度达到95%以上。

它们唯一的区别就是当在自旋过程中线程被阻塞到指定被唤醒时，如果此时线程的重点标志位是true，前者会抛出InterruptedException中断异常，后者则只会将自己的中断标志设置为true，不会抛出异常。

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

3.1.3 tryAcquireSharedNanos方法源码剖析

tryAcquireSharedNanos方法则是一个加入了最大等待时间的入队方法，它的主流程也是和前面两个差不多。

这个最大等待时间是发挥什么作用的呢？其实它是当自旋超过N次还获取不到锁的时候，阻塞线程的时间是有限制的，即拥有一个最大的阻塞时间，这个时间就是传入的最大等待时间-自旋消耗的总时间。

如果在没有进入阻塞线程的方法时，自旋时间已经超过了最大等待时间，就会直接返回false退出整个自旋操作。同时，在自旋中，如果当前线程的中断标志为true也会抛出中断异常。

    private boolean doAcquireSharedNanos(int arg, long nanosTimeout)throws InterruptedException {if (nanosTimeout <= 0L)return false;final long deadline = System.nanoTime() + nanosTimeout;final Node node = addWaiter(Node.SHARED);boolean failed = true;try {for (;;) {final Node p = node.predecessor();if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return true;}}nanosTimeout = deadline - System.nanoTime();// 如果还没阻塞的时候，自旋时间已经超过最多时间了直接返回if (nanosTimeout <= 0L)return false;if (shouldParkAfterFailedAcquire(p, node) &&nanosTimeout > spinForTimeoutThreshold)// 不会无期限阻塞，会有一个最多阻塞时间LockSupport.parkNanos(this, nanosTimeout);if (Thread.interrupted())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}}

3.1.4 共享锁入队总结

共享锁能同时被多个线程持有，也存在并发释放锁的过程，因此会有一个传播行为，即不仅仅释放当前等待节点，还会通知后面的节点。

获取锁的方法从阻塞角度可分为两类：

①无指定阻塞时间的，最多自旋两次还拿不到锁就将线程阻塞，直到等待其他持有锁的线程释放锁以后才能唤醒阻塞在这里的线程。
②有指定最大等待时间的，只要拿不到锁就会将线程阻塞在最多是最大等待时间内，这保证了当前线程不会被一直阻塞。

根据这两大特点，是不是可以用AQS的共享锁来做类似流量控制的操作以及等待多任务执行的全部完成呢？

的确，在AQS共享锁的子类中，LimitLatch就是应用于流量控制，它被tomcat的NIO非阻塞模型使用。CountDownLatch则通常被应用于等待完成操作。

3.2 共享锁出队

共享锁的出队和独占锁的出队不一样，独占锁的出队只会唤醒一次头节点的后继节点，而共享锁则会在一定情况下通过死循环不断的通知。

AQS#releaseShared(int arg)是释放共享锁的实现，它的逻辑分两部分：

调用子类重写的钩子方法tryReleaseShared(int arg)，返回true表示当前线程持有的共享锁已经释放了。false则表示此时还无法释放共享锁。
当共享锁释放成功后，调用doReleaseShared()方法，这个是通知、唤醒等待节点的底层实现，即当前线程释放了共享锁，还要通知后面等待着的节点们。

    public final boolean releaseShared(int arg) { // 释放共享锁if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;}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))continue;            // loop to recheck cases// 唤醒头节点的有效后继节点（实际的等待节点）unparkSuccessor(h);}else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue;                // loop on failed CAS}// 如果经过unparkSuccessor(h)方法时，头节点被改变了// 则继续自旋通知最新头节点后面的节点if (h == head)                   break;}}

doReleaseShared()方法是一个自旋操作，每次自旋时都会拿到最新的头节点，如果不为null并且队列不为空（为空时head=tail）时，根据头节点的waitStatus做判断：

如果是SIGNAL表示需要信号，那么这就是说明此时队列有等待节点（绑定的线程也在阻塞中），SIGNAL是在doAcquireShared...方法自旋重新竞争锁时的预判阻塞操作中设置的。然后通过CAS将状态从SIGNAL转变为0，表示从信号转为处理中——unparkSuccessor(h)方法就是找到传入节点（头节点）最近的一个有效的等待节点将其从阻塞状态中唤醒。
如果waitStatus=0，则会通过CAS操作将其设置为PROPAGATE（-3），这个设置的意义是什么呢？还记得在setHeadAndPropagate中的传播行为条件吗？其中一个条件就是此时的头节点的waitStatus<0。那么回到这里设置为PROPAGATE的目的就是为了保持传播行为。

退出自旋的条件是经过unparkSuccessor(h)方法后，头节点还没有被改变，即唤醒的等待节点重新进入doAcquireShared...自旋竞争锁还是失败了，没有升级为头节点。

独占锁模式下的释放锁

    public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}

doReleaseShared()自旋的目的，笔者认为还是和共享锁能同时被多个线程持有相关，既然能同时持有，也能同时释放，这样做的目的是能在最短时间通知更多的等锁节点去重新竞争共享锁。

而独占锁模式下的释放锁，没有自旋，仅仅是通知一个等待节点。因为锁的获取和释放是一对一的接力关系，即使你通知更后面的等待节点也没用，也必须被动等待释放锁的线程通知。

4.案例

经过3.1 共享锁入队分析，我们知道了共享锁是可以多个线程持有的、并且从阻塞角度看，共享锁入队时会造成等待线程的无期限阻塞和有时间限制的阻塞。

因此基于这两个阻塞特点，可以利用共享锁去用作等待多任务执行完成的情况。

以下面的CountDownLatch为例子，下面的是典型的等待多任务完成的例子，创建一个门闩数量是3的CountDownLatch实例，并且构造三个线程并传入此CountDownLatch实例，当睡眠完成后，调用countDown()方法将此时的门闩数减一。

最后在main函数的主线程中调用await()等待三个线程都完成。

public class CountDownLatchTest {volatile static int completeSize = 0;public static void main(String[] args) throws InterruptedException {CountDownLatch countDownLatch = new CountDownLatch(3);for (int i=0; i<3; i++) {int baseSleepTime = 2000;if (i==1){baseSleepTime = 3 * baseSleepTime;}new Thread(new RunnableThread(countDownLatch,(baseSleepTime * i), i)).start();}countDownLatch.await();if (completeSize==3){System.out.println("所有任务都完成了");}}static class RunnableThread implements Runnable {private int sleepTime;private int num;private CountDownLatch countDownLatch;// ...构造函数@Overridepublic void run() {try {System.out.println("RunnableThread"+num+"：开始睡眠");Thread.sleep(sleepTime);countDownLatch.countDown();completeSize++;System.out.println("RunnableThread"+num+"：完成执行");}catch (Exception e){}}}
}
-----------------输出结果---------------------------
RunnableThread2：开始睡眠
RunnableThread0：开始睡眠
RunnableThread1：开始睡眠
RunnableThread0：完成执行
RunnableThread2：完成执行
RunnableThread1：完成执行
所有任务都完成了
-----------------输出结果---------------------------

5.总结

共享锁和独占锁，并不是字面意思上的对立，而是共享锁是独占锁的一个补充，共享锁同时能被多个线程持有，而独占锁同一时刻只能由一个线程持有。

因此共享锁模式下的入队和出队和独占锁模式的会有差异：

入队中的差异主要在于在自旋过程中、等待节点二次获取到锁后除了升级为头节点外、还多了一个传播行为——因为，共享锁是可以同时持有，也可以同时被释放，因此就需要多做一步通知、唤醒后面的等待节点。
出队中的差异在于共享锁模式下也是自旋的出队，每次自旋都会拿到最新头节点，尝试将该头节点后面的等待节点唤醒，自旋退出条件是唤醒的等待节点没有重新获取锁升级为新头节点。自旋的目的和入队中一样，也是因为共享锁是可以同时持有，也可以同时被释放，即这么做能在最短时间内通知更多的等待节点去获取锁。

在共享锁入队的章节中分析的三个入队方法，也得知其实共享锁可以应用于流量控制、等待多任务完成的背景。

实际上，LimitLatch的确被tomcat使用到了，其用途也是类似于流量控制等。CountDownLatch则是用来等待多任务的完成。