AbstractQueuedSynchronizer源码分析

前提

AQS（java.util.concurrent.locks.AbstractQueuedSynchronizer）是并发编程大师Doug Lea创作的用来构建锁或者其他同步组件（信号量、事件等）的基础框架类。在J2SE 1.5的java.util.concurrent包（下称JUC包）中，大部分的同步器（例如锁，屏障等等）都是基于AbstractQueuedSynchronizer类（下称AQS类）这个简单的框架(说实话，作为一个数据结构和算法比较一般的人来说阅读其源码真的一点也不简单)而构建的。这个框架为同步状态的原子性管理、线程的阻塞和解除阻塞以及排队提供了一种通用的机制。由此可知，学习AQS的源码对理解JUC包的其他类的功能和实现有极大的帮助。AQS在设计的时候主要使用了模板方法这一设计模式，使用者只需要继承AQS此抽象类，并且重写指定的方法，那么在并发组件的实现中就能调用AQS的模板方法，模板方法会调用使用者覆写的方法。AQS的主要子类如下:

ReentrantLock、ReentrantReadWriteLock、Semaphore中的NonfairSync和FairSync。
CountDownLatch。
ThreadPoolExecutor中的Worker。

主要参考资料：

自旋锁、排队自旋锁、MCS锁、CLH锁
AbstractQueuedSynchronizer源码解读
java.util.concurrent.locks.AbstractQueuedSynchronizer源码
JUC同步器框架
一行一行源码分析清楚AbstractQueuedSynchronizer

预备知识

CLH队列

CLH队列其实是CLH锁的一种变形，思路是大致相同，数据结构和机制有调整。调整点：

节点的结构调整：引入了头结点和尾节点，它分别指向队列的头和尾，尝试获取锁、入队列、释放锁等实现都与头尾节点相关，并且每个节点都引入前驱节点和后继节点的引用，每个节点单独保存节点的状态。
节点的等待(唤醒)机制调整：CLH锁中的等待机制是在前驱节点的属性上自旋，而在AQS的CLH队列中自旋增加了特性以前驱节点是否为head决定阻塞唤醒。

CLH锁相关应用可以看参考资料里面的那篇文章。值得注意的是AQS中并没有引入Queue的实现作为队列，而是只使用了两个字段head和tail来维护一个链表队列，这两个字段是可原子更新的，两者在初始化时都指向了一个空节点，并且该链表队列是FIFO（先进先出）的。

阻塞和唤醒

节点的阻塞和唤醒依赖到LockSupport这个类，里面用到的是Unsafe的相关方法，之前写过一篇介绍Unsafe的文章，有兴趣可以看下：JAVA中神奇的双刃剑--Unsafe。

AQS的实现原理

AQS本质是同步器，同步器一般包含两种方法，一种是acquire，另一种是release。acquire操作阻塞调用的线程，直到或除非同步状态允许其继续执行。而release操作则是通过某种方式改变同步状态，使得一或多个被acquire阻塞的线程继续执行。同步器的实现根据其状态是否独占而有所不同。独占状态的同步器，在同一时间只有一个线程可以通过阻塞点，而共享状态的同步器可以同时有多个线程在执行。一般锁的实现类往往只维护独占状态，但是，例如计数信号量在数量许可的情况下，允许多个线程同时执行。为了使框架能得到广泛应用，这两种模式都要支持。

AQS内部维护一个CLH队列的变体来管理锁。线程会首先尝试获取锁，如果失败，则将当前线程以及等待状态等信息包成一个Node节点加到同步队列的队尾。接着会不断循环尝试获取锁（条件是当前节点为head的直接后继节点才会尝试，这样做可以防止大量节点向前驱节点自旋造成大量不必须的性能消耗），如果失败则会阻塞自己，直至被唤醒；而当持有锁的线程释放锁时，会唤醒队列中的后继线程。

同步器背后的基本思想非常简单。acquire操作如下：

//这个是论文里面的原文

while (synchronization state does not allow acquire) {

    enqueue current thread if not already queued;

    possibly block current thread;

}

dequeue current thread if it was queued;

//翻译如下

while(同步状态不允许其他线程获取到锁){

    把当前线程包装成节点插入同步队列(的队尾)

    if(需要阻塞当前线程){

        阻塞当前线程直至被唤醒

    }

}

release操作如下：

//这个是论文里面的原文

update synchronization state;

if (state may permit a blocked thread to acquire)

    unblock one or more queued threads;

//翻译如下

更新同步状态status的值

if(修改后的同步状态允许其他线程获取到锁){

    唤醒一个或者多个队列中的等待线程

}

为了实现上述操作，需要下面三个基本组件的相互协作：

1、同步状态的原子性管理。
2、线程的阻塞与解除阻塞。
3、同步队列的管理。

1. 同步状态的原子性管理

AQS中提供三个protected final方法是AQS中用来访问/修改同步状态的方法:

int getState(): 获取同步状态。
void setState(): 设置同步状态。
boolean compareAndSetState(int expect, int update)：基于CAS，原子设置同步状态。

2. 线程的阻塞与解除阻塞

基于java.util.concurrent.locks.LockSupport的park和unpark方法对线程进行阻塞和解除阻塞，底层依赖sun.misc.Unsafe的本地方法park和unpark(暂时没有能力从JVM层面研究)。

3. 同步队列的管理

见下面的数据结构分析和方法源码分析。

AQS主要数据结构

Node结构

在AQS的等待队列中，每一个线程被包装为一个Node实例，数据结构是链表，内部类Node的源码如下：

static final class Node {

        // 标记节点当前在共享模式下

        static final Node SHARED = new Node();

        // 标记节点当前在独占模式下

        static final Node EXCLUSIVE = null;

        /****************下面这四个常量是waitStatus的候选值******************/

        // 等待状态：取消。代表此节点因为超时或者中断被取消，表现为被取消的节点中的线程会跳出阻塞(简单来说就是放弃抢占锁)

        static final int CANCELLED =  1;

        // 等待状态：唤醒。代表此节点的后继节点需要被唤醒

        static final int SIGNAL    = -1;

        // 等待状态：CONDITION(不知道怎么翻译，叫条件好像也不合理，干脆用英文)。代表此节点在Condition队列中，此状态下节点不会作为同步队列的节点，除非状态跃迁为0。

        static final int CONDITION = -2;

        // 等待状态：传播。用于将唤醒后继节点传递下去，这个状态的引入是为了完善和增强共享锁的唤醒机制。在一个节点成为头节点之前，是不会跃迁为此状态。

        static final int PROPAGATE = -3;

        //等待状态，可选值为-3,-2,-1,0,1

        volatile int waitStatus;

        // 前驱节点的引用

        volatile Node prev;

        // 后继节点的引用

        volatile Node next;

        // 当前节点对应的线程

        volatile Thread thread;

        // 共享模式下或者在CONDITION中后继节点的引用

        Node nextWaiter;

        // 检查是否SHARED(共享)模式

        final boolean isShared() {

            return nextWaiter == SHARED;

        }

        // 获取前驱节点，如果为null抛出NullPointerException异常

        final Node predecessor() throws NullPointerException {

            Node p = prev;

            if (p == null)

                throw new NullPointerException();

            else

                return p;

        }

        // 使用此构造新建的Node实例将会作为等待队列的虚拟头节点或者共享模式下的标记

        Node() {    // Used to establish initial head or SHARED marker

        }

        // addWaiter会调用此构造函数

        Node(Thread thread, Node mode) {     // Used by addWaiter

            this.nextWaiter = mode;

            this.thread = thread;

        }

        // Condition会用到此构造函数

        Node(Thread thread, int waitStatus) { // Used by Condition

            this.waitStatus = waitStatus;

            this.thread = thread;

        }

    }

为了部分场景的描述，等待队列、同步队列和CLH队列都是指AQS中的Node等待队列。

这里可以从Node的结构做一下笔记：

1、AQS的等待队列中除了头节点外，节点的前驱节点必定不为null。
2、waitStatus初始状态为0，只要waitStatus大于0说明节点被取消，即线程放弃抢占锁，将会释放原来的阻塞等待状态。
3、AQS的CLH队列中，头节点是虚拟的，也就是头节点不包含线程的引用。

如果不考虑共享模式和CONDITION的使用，Node 的数据结构其实也挺简单的，就是thread + waitStatus + pre + next四个属性而已。

Node状态迁移

上面的Node数据结构提到等待状态waitStatus的取值问题，总结如下：

状态值	解释
0	初始状态，也就是默认值
1	取消。代表此节点因为超时或者中断被取消被取消，表现为被取消的节点中的线程会跳出阻塞(简单来说就是放弃抢占锁)
-1	唤醒。代表此节点的后继节点需要被唤醒
-2	CONDITION。代表此节点在Condition队列中，此状态下节点不会作为同步队列的节点，除非状态跃迁为0
-3	传播。用于将唤醒后继节点传递下去，这个状态的引入是为了完善和增强共享锁的唤醒机制。在一个节点成为头节点之前，是不会跃迁为此状态

这里晚点补充一张状态跃迁的图，因为还没有细致研究过状态跃迁的过程。虽然参考资料里面有涉及到，但是还没实践过。

AQS基本属性

// 头结点

private transient volatile Node head;

// 尾节点

private transient volatile Node tail;

// 同步状态

private volatile int state;

// 代表当前持有独占锁的线程

private transient Thread exclusiveOwnerThread;

head：表示AQS等待队列的头节点引用，它是延迟初始化的。head在逻辑上的含义是当前持有锁的线程，但实际上head节点是一个虚节点，本身并不会存储线程信息(这里可能有点绕，下面分析方法代码的时候再提一下)。
tail：表示AQS等待队列的头节点引用，它是延迟初始化的。当一个线程无法获取锁而被加入到同步队列时，会用CAS来设置尾节点tail为包装当前线程的Node节点(简单理解为新入队的Node就是tail)。
state：同步状态，初始值为0。代表的是当前锁的状态，0表示没有被占用，大于0代表有线程持有当前锁，每重入一次此值加1。对于信号量来说，它代表当前已经被占用的信号量。
exclusiveOwnerThread：代表当前持有独占锁的线程引用。因为锁可以重入，需要通过当前锁持有的线程去判断重入的次数，举个例子，ReentrantLock#lock可以多次调用，伪代码如下：

if (currentThread == getExclusiveOwnerThread()) {state++;}

exclusiveOwnerThread属性存放在AQS的父类AbstractOwnableSynchronizer，这个类只包含exclusiveOwnerThread属性和它的getter、setter方法。

理解这些基本属性和Node的数据结构之后，可以大概画出AQS中CLH队列(存在等待节点时候)的模型：

这里的模型忽略了节点中持有的线程引用或者共享模式下的waiter引用。

AQS的主要方法源码解读

AQS提供下面几个protected方法交由子类覆盖，用于实现同步状态的管理，这些方法默认实现都是直接抛出异常的，因此子类必须覆盖才能正常使用：

方法	方法说明
boolean tryAcquire(int arg)	尝试获取独占锁
boolean tryRelease(int arg)	尝试释放独占锁
int tryAcquireShared(int arg)	尝试获取共享锁
boolean tryReleaseShared(int arg)	尝试释放独占锁
boolean isHeldExclusively()	当前线程是否获得了独占锁

以上的几个试获取/释放锁的方法的具体实现应当是无阻塞的。

AQS本身将同步状态的管理用模板方法模式都封装好了，以下列举了AQS中的一些模板方法，它们都是public修饰的，内部依赖了上面提到的实现同步状态的管理的由子类实现的方法：

方法	方法说明
void acquire(int arg)	获取独占锁。会调用`tryAcquire`方法，如果未获取成功，则会进入同步队列等待
void acquireInterruptibly(int arg)	响应中断的`acquire`方法版本
boolean tryAcquireNanos(int arg,long nanos)	响应中断+带有超时时间的`acquire`方法版本
void acquireShared(int arg)	获取共享锁。会调用`tryAcquireShared`方法
void acquireSharedInterruptibly(int arg)	响应中断的`acquireShared`方法版本
boolean tryAcquireSharedNanos(int arg,long nanos)	响应中断+带有超时时间的`acquireShared`方法版本
boolean release(int arg)	释放独占锁
boolean releaseShared(int arg)	释放共享锁
Collection getQueuedThreads()	获取同步队列上的线程集合

这些模板方法主要区分不同的维度：获取和释放、独占和共享、支持响应中断和不支持响应中断。如果有细心留意到这些方法的arg参数都是int类型，其实这个就是传入的目标state的更新值。

下面重点分析一下获取独占锁、释放独占锁、获取共享锁和释放共享锁的源码实现。

获取独占锁

独占锁只允许一个线程持有锁，后续申请持有锁的线程都会阻塞，直到锁被释放。获取独占锁最典型的方法就是ReentrantLock中的lock方法，下面我们就以此作为源码分析，先看ReentrantLock的静态内部类FairSync：

    static final class FairSync extends Sync {

        private static final long serialVersionUID = -3000897897090466540L;

        final void lock() {

            acquire(1);

        }

        protected final boolean tryAcquire(int acquires) {

            final Thread current = Thread.currentThread();

            int c = getState();

            //state = 0，说明没有线程抢占锁，处于初始化状态

            if (c == 0) {

                if (

                    //没有其他线程占用CLH队列的头节点的后继节点

                    !hasQueuedPredecessors() &&

                    //CAS设置state，实际上就是原子更新state从0更新为1

                    compareAndSetState(0, acquires)) {

                    //设置独占线程引用为当前线程

                    setExclusiveOwnerThread(current);

                    return true;

                }

            }

            //state大于0(这里肯定不会小于0，因为在这个方法里面只会更新为acquires)，判断当前线程是否和AQS的独占线程相同，如果相同则说明当前线程锁重入，则设置同步状态为原值+acquires(在这里实际是state = state + 1)

            else if (current == getExclusiveOwnerThread()) {

                int nextc = c + acquires;

                if (nextc < 0)

                    throw new Error("Maximum lock count exceeded");

                setState(nextc);

                return true;

            }

            return false;

        }

    }

    //AbstractQueuedSynchronizer#hasQueuedPredecessors

    //这个方法返回true的条件是：

    //1、头节点和尾节点不相等。

    //2、头节点的后继节点为null或者头节点的后继节点的持有线程不等于当前线程。

    //简单来说就是：当前的线程不在等待队列的头节点的直接后继节点，有其他线程占用该节点的时候返回true。返回false说明可以把当前线程包装到新Node作为头节点的后继节点。

    public final boolean hasQueuedPredecessors() {

        // The correctness of this depends on head being initialized

        // before tail and on head.next being accurate if the current

        // thread is first in queue.

        Node t = tail; // Read fields in reverse initialization order

        Node h = head;

        Node s;

        return h != t &&

            ((s = h.next) == null || s.thread != Thread.currentThread());

    }

接下来可以看AbstractQueuedSynchronizer#acquire，它依赖到上面分析到的tryAcquire方法：

    public final void acquire(int arg) {

        if (

            //如果获取到锁就直接返回

            !tryAcquire(arg)

            && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

            //就是调用Thread.currentThread().interrupt()，也就是自身中断

            selfInterrupt();

    }

    //添加新的节点到CLH队列中，这里的mode为Node.EXCLUSIVE，实际值为null

    private Node addWaiter(Node mode) {

        /**

         * 这里使用的构造函数是：

         * Node(Thread thread, Node mode) {

         *   this.nextWaiter = mode;

         *   this.thread = thread;

         *  }

         * 注意到nextWaiter为null。

         */

        Node node = new Node(Thread.currentThread(), mode);

        //快速尝试一次入队操作，如果失败降级到enq方法

        Node pred = tail;

        //尾节点为null，说明还没有初始化，如果不为null，则已经初始化

        if (pred != null) {

            //当前节点的前驱节点引用设置为CLH队列队尾的节点

            node.prev = pred;

            //通过CAS在队尾插入当前节点

            if (compareAndSetTail(pred, node)) {

                pred.next = node;

                return node;

            }

        }

        //尾节点未初始化或者快速尝试一次入队操作失败后会降级到此方法

        enq(node);

        return node;

    }

    private Node enq(final Node node) {

        for (;;) {

            Node t = tail;

            //尾节点为null，说明CLH队列中的头节点和尾节点都没有初始化

            if (t == null) { // Must initialize

                //CAS设置一个虚拟的头节点

                if (compareAndSetHead(new Node()))

                    //尾节点赋值为虚拟的头节点

                    tail = head;

            } else {

                //尾节点不为null，说明头节点和尾节点已经初始化

                //当前节点的前驱节点赋值为尾节点

                node.prev = t;

                //CAS设置尾节点为当前的节点

                if (compareAndSetTail(t, node)) {

                    //CAS设置尾节点为当前的节点成功后，设置上一个尾节点的后继节点为当前节点

                    t.next = node;

                    return t;

                }

            }

        }

    }

    /**

     * 方法的名称是"获取入队列"，实际上这个方法里面就是自旋获取锁的过程

     */

    final boolean acquireQueued(final Node node, int arg) {

        //默认是获取锁失败

        boolean failed = true;

        try {

            //默认为非中断状态

            boolean interrupted = false;

            for (;;) {

                final Node p = node.predecessor();

                /**

                 * 检测当前节点的前驱是否head，这是试获取锁的前提条件。

                 * 如果满足前节点的前驱是head，调用tryAcquire尝试获取锁。

                 * 如果获取锁成功，则设置head为当前节点，并且把head的后继节点置为null。

                 */

                if (p == head && tryAcquire(arg)) {

                    setHead(node);

                    p.next = null; // help GC

                    failed = false;

                    return interrupted;

                }

                /**

                 * 走到这里说明了当前节点的前驱节点不是head或者尝试获取锁失败了。

                 * 通过shouldParkAfterFailedAcquire判断是否需要阻塞当前线程。

                 * parkAndCheckInterrupt用于阻塞当前线程，并且返回当前线程的中断状态，如果线程被中断而跳出阻塞，interrupted设置为true。

                 */

                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())

                    interrupted = true;

            }

        } finally {

            if (failed)

                cancelAcquire(node);

        }

    }

    // 通过当前节点以及当前节点的前驱节点判断是否需要阻塞当前线程

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

        int ws = pred.waitStatus;

        if (ws == Node.SIGNAL)

            /*

             * This node has already set status asking a release

             * 前驱节点已经被设置为SIGNAL状态，意味着释放锁的时候会唤醒后继(当前)节点，所以可以安全地阻塞当前的线程。

             * to signal it, so it can safely park.

             */

            return true;

        if (ws > 0) {

            /*

             * Predecessor was cancelled. Skip over predecessors and

             * indicate retry.

             * 前驱节点的状态大于0(只有一个状态是大于0的，就是cancelled)，则不断地向前遍历，得到一个非取消的前驱节点作为当前节点的前驱节点。

             * 非取消的前驱节点的后继节点设置为当前节点。

             * 此步骤结束后会进入下一轮循环，也就是再次尝试获取锁。

             */

            do {

                node.prev = pred = pred.prev;

            } while (pred.waitStatus > 0);

            pred.next = node;

        } else {

            /*

             * waitStatus must be 0 or PROPAGATE.  Indicate that we

             * need a signal, but don't park yet.  Caller will need to

             * retry to make sure it cannot acquire before parking.

             * 剩下的等待状态的值一定是0或者PROPAGATE(-3)，CAS设置前驱的等待状态为SIGNAL，此步骤结束后会进入下一轮循环，

             * 也就是再次尝试获取锁(不进行线程的阻塞)。

             *

             */

            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);

        }

        return false;

    }

    /**

     * 调用LockSupport的park方法对当前线程进行阻塞，返回当前线程的中断状态(返回后清除中断标记)

     */

    private final boolean parkAndCheckInterrupt() {

        LockSupport.park(this);

        return Thread.interrupted();

    }

    /**

     * 取消指定node获取锁的操作，实际上就是指定node出队的过程

     */

    private void cancelAcquire(Node node) {

        // Ignore if node doesn't exist

        if (node == null)

            return;

        node.thread = null;

        // 遍历并更新节点前驱，把当前节点的prev指向队列前方第一个非CANCELLED的节点。

        Node pred = node.prev;

        while (pred.waitStatus > 0)

            node.prev = pred = pred.prev;

        // predNext is the apparent node to unsplice. CASes below will

        // fail if not, in which case, we lost race vs another cancel

        // or signal, so no further action is necessary.

        /**

         * 这里的pred是指当前节点的直接非取消的前驱节点。

         * predNext记录了这个前驱节点原来指向的后继节点(不一定是当前节点)，下面的CAS操作会用到这个节点。

         */

        Node predNext = pred.next;

        // Can use unconditional write instead of CAS here.

        // After this atomic step, other Nodes can skip past us.

        // Before, we are free of interference from other threads.

        /**

         * 当前节点的等待状态直接设置为取消，而不是使用CAS。

         * 此原子步骤执行后，其他节点的更新操作可以跨越此节点。

         * 在这之前，其他线程不会干扰到此节点的状态操作。

         */

        node.waitStatus = Node.CANCELLED;

        /**

         * 如果当前节点就是尾节点，通过CAS直接把前驱节点pred设置为尾节点。

         * 如果CAS更新前驱节点为尾节点成功，则需要把前驱节点的后继节点引用(preNext)设置为null，这样做就彻底断开了pred和其他后继节点(包括node的关联)，

         * 这些被断开的后继节点相当于全部出队，最终被回收。

         * 如果这里涉及多线程操作，总有一个线程会更新成功，只要有一个成功就可以了，链表队列就能重建。

         */

        if (node == tail && compareAndSetTail(node, pred)) {

            compareAndSetNext(pred, predNext, null);

        } else {

            // If successor needs signal, try to set pred's next-link

            // so it will get one. Otherwise wake it up to propagate.

            /**

             * 这里是当前节点不是尾节点或者CAS设置尾节点为pred失败就会进来else逻辑(说实话，下面的if太复杂有点吓人)

             * 如果前驱节点pred不为头节点，并且pred的等待状态为SIGNAL或者非取消状态并且成功设置为SIGNAL，并且pred持有的线程引用不为null，

             * 如果当前节点的直接后继节点不空并且不为取消状态，则用CAS尝试把pred的后继引用置为node的直接后继节点。

             * 如果这里涉及多线程操作，总有一个线程会更新成功，只要有一个成功就可以了，链表队列就能重建。

             */

            int ws;

            if (pred != head &&

                ((ws = pred.waitStatus) == Node.SIGNAL ||

                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&

                pred.thread != null) {

                Node next = node.next;

                if (next != null && next.waitStatus <= 0)

                    compareAndSetNext(pred, predNext, next);

            } else {

                /**

                 * 从上面的if分支可以知道，走到这里说明pred == head，pred的状态为取消或者pred.thread == null。

                 * 1.pred == head，当前节点的直接前驱节点为head，说明这个时候应该唤醒后继(当前)节点(如果阻塞)去获取锁，否则队列管理出现问题。

                 * 2.pred的状态为取消，这样显然要唤醒后继节点。

                 * 3.类似情况1。

                 * 此方法没有更新pred和next的关系，有可能导致pred节点后的队列节点中存在大量的处于取消状态的节点。

                 * 被唤醒的直接后继节点尝试获取锁的时候，如果节点处于取消状态，尝试获取锁(tryAcquire)失败，

                 * 会走进去shouldParkAfterFailedAcquire逻辑，在里面会维护当前节点的pred和next，踢掉这些取消的节点。

                 */

                unparkSuccessor(node);

            }

            /**

             * 这里有个疑惑，为什么把当前取消的节点的next引用设置为自身，注释写着：help GC？

             *

             * 摘抄一下参考资料里面的内容说：

             * 为了方便AQS中Condition部分的isOnSyncQueue方法，判断一个原先属于条件队列的节点是否转移到了同步队列。

             * 因为同步队列中会用到节点的next域，取消节点的next也有值的话，可以断言next域有值的节点一定在同步队列上。

             *

             * 在GC层面，和设置为null具有相同的效果。

             */

            node.next = node; // help GC

        }

    }

    /**

     * 唤醒后继节点中的线程

     */

    private void unparkSuccessor(Node node) {

        /*

         * If status is negative (i.e., possibly needing signal) try

         * to clear in anticipation of signalling.  It is OK if this

         * fails or if status is changed by waiting thread.

         */

         //如果等待状态为负数(后继节点有可能需要被唤醒)，通过CAS重置状态值为0，从而让后继节点可以重新尝试获取锁。

        int ws = node.waitStatus;

        if (ws < 0)

            compareAndSetWaitStatus(node, ws, 0);

        /*

         * Thread to unpark is held in successor, which is normally

         * just the next node.  But if cancelled or apparently null,

         * traverse backwards from tail to find the actual

         * non-cancelled successor.

         */

        /**

         * 需要被解除阻塞的线程在当前节点的后继节点中，通常就是后一个(直接后继)节点。但是如果后继节点为null或者被取消，

         * 这里必须找到一个真正的非null并且非取消的后继节点。

         *

         * 这里的逻辑主要是从尾节点开始，向前驱节点做遍历(遍历的上限时当前节点)，寻找一个节点的等待状态是小于等于0(非取消状态)，赋值给s。

         * 注意可能有以下的情形：

         * 1、当s == null，也就是当前节点的后继节点为null，当前节点不一定是tail，也有可能是正在有新的节点入队。

         * 2、假设s == null，并且当前的node就是tail。

         * 3、通过Node s = node.next读取s的时候确实s == null，下一刻node节点之后已经添加新节点，s变为非null。

         */

        Node s = node.next;

        if (s == null || s.waitStatus > 0) {

            s = null;

            for (Node t = tail; t != null && t != node; t = t.prev)

                if (t.waitStatus <= 0)

                    s = t;

        }

        // 搜索到的前驱节点不为null，则唤醒其中的线程

        if (s != null)

            LockSupport.unpark(s.thread);

    }

简单总结acquire的流程是：

1、首先尝试获取一次锁，如果成功，则返回。
2、第一次获取锁失败，把当前线程包装成Node插入到CLH队列中，在队列中会检测是否为head的直接后继，并尝试获取锁。
3、如果当前线程所在的节点不是head的直接后继，通过LockSupport阻塞当前线程，直至被释放锁的线程唤醒或者被中断，随后再次尝试获取锁，如此反复。

注意点一

注意到源码里面的enq方法的else分支逻辑，前驱节点的设置总是在CAS操作之前，CAS操作用于设置尾节点，而后继节点的设置在尾节点设置成功后。这里的逻辑是在原来的CLH队列中的队尾添加一个新节点，但是为什么代码逻辑是这样的，代码逻辑的顺序调乱不行吗？这就是AQS的代码的细节和严谨之处。

注意点二

注意到acquireQueued方法的finally块里面的cancelAcquire方法的执行时机。当时晚上看源码，脑子不灵活想了很久没想懂，第二天早上在stackoverflow提了个问。其实很简单，try-finally不管抛出异常或者正常返回都会执行。在acquireQueued方法中如果正常返回，在return之前总是会把failed设置为false，因此不会进去cancelAcquire方法。也就是当逻辑进入到cancelAcquire方法，必定是抛出了异常，因为没有显式的catch块代码，因此异常必定是非检查型(unchecked)的异常，在这里最大可能的就是RuntimeException的子类。查看整个方法，只有Node#predecessor有可能抛出NullPointerException，也就是当前节点的前驱节点为null的时候(这里暂时分析不出来在什么情况下，当前节点的前驱节点是null)。

释放独占锁

接着看释放独占锁的方法，主要是AQS的release方法，依赖到子类需要覆盖的方法tryRelease，这里还是分析一下典型例子ReentrantLock内部类Sync的tryRelease方法。

    //ReentrantLock$Sync#tryRelease

    protected final boolean tryRelease(int releases) {

        //AQS状态值减去当前传入的releases(这里是1)

        int c = getState() - releases;

        // 如果当前锁独占的线程不是当前线程，抛出异常，这里就是A线程加锁，B线程解锁的场景

        if (Thread.currentThread() != getExclusiveOwnerThread())

            throw new IllegalMonitorStateException();

        boolean free = false;

        //只有state衰减到0，设置AQS的独占线程为null，并且设置free为true，也就是会返回true

        if (c == 0) {

            free = true;

            setExclusiveOwnerThread(null);

        }

        //更新state的值

        setState(c);

        return free;

    }

    /**

     * AQS#release:释放锁

     */

    public final boolean release(int arg) {

        //先调用上面提到的子类的tryRelease方法，此方法只有当state值为0的时候才返回true

        if (tryRelease(arg)) {

            Node h = head;

            // head节点的状态不可能为取消，因为它不持有线程引用，因此不等于0，说明小于0。

            if (h != null && h.waitStatus != 0)

                // 唤醒后继节点中的线程，此函数在acquire中已经分析过，不再列举说明

                unparkSuccessor(h);

            return true;

        }

        return false;

    }

最后分析一下，release方法中，通过Node h = head，这里分析一下h也就是头节点会有多少种可能的值：

1、head为null，也就是AQS刚初始化的情况。
2、head不为null，由于acquireQueued方法有新的节点插入到CLH队列尾部，通过setHead设置head节点。
3、head不为null，由于head的直接后继节点释放了锁，有新的非取消的后继节点被唤醒在acquireQueued的新一次循环中尝试成功通过setHead设置head节点。

release方法只会在子类需要覆盖的方法tryRelease返回为true，并且头节点不空并且头节点状态小于0的情况下才会唤醒后继节点。

图解获取和释放独占锁

下面试图总结一下获取和释放独占锁的过程。

单线程重入

为了简单说明问题，使用了ReentrantLock的公平锁做debug，代码例子如下：

    public static void main(String[] args) throws Exception{

        ReentrantLock lock = new ReentrantLock(true);

        lock.lock();

        lock.lock();

        lock.unlock();

        lock.unlock();

    }

第一个lock.lock()调用后，单线程不存在竞争，没有元素入队，所以head和tail都不会初始化，如下：

注意，因为锁重入不存在竞争tryAcquire总是成功的，所以不需要新节点入队，因此第二个lock.lock()调用后，如下：

接下来释放锁也很简单，第一个lock.unlock()调用后(值得注意的是，这一步的sync.release(1)是返回false的)，如下：

第二个lock.unlock()调用后(值得注意的是，这一步的sync.release(1)是返回true的)，如下：

单线程锁重入的不存在抢占或者等待，所以是比较简单的，这里的head和tail是不存在的，因为head和tail由头到尾都没有初始化。

多线程抢占

为了简单说明问题，使用了ReentrantLock的公平锁做debug，这里只模拟两个线程的情况，更多线程也是类似的，代码例子如下：

    public static void main(String[] args) throws Exception{

        ReentrantLock lock = new ReentrantLock(true);

        Runnable runnable = () -> {

            lock.lock();

            System.out.println("pass");

            lock.unlock();

        };

        Thread thread1 = new Thread(runnable);

        thread1.setName("thread-1");

        Thread thread2 = new Thread(runnable);

        thread2.setName("thread-2");

        thread1.start();

        thread2.start();

        TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);

    }

如果使用IDEA的话，把debug模式由All改为选择Thread。因为线程调度具有不确定性，现在假定"thread-2"先运行进入Runnable实例调用lock.lock()，没有多线程抢占需要入队的情况下，head和tail不会初始化:

接着hold住断点在tryAcquire方法返回之前，切换到"thread-1"放行，"thread-1"运行进入Runnable实例调用lock.lock()，然后观察变量的更变值：

此时，线程"thread-1"被阻塞了。记住，上面的断点入口要打在lock.lock()和lock.unlock()，断点的出口要打在 System.out.println("pass")。接着切换回去"thread-2"进行解锁，注意此时"thread-1"被唤醒重新获得锁，head和tail由于已经初始化，状态被重置，CLH队列中没有实际节点元素，tail处于游离状态。

"thread-1"解锁。

获取共享锁

与获取独占锁的实现不同，共享锁允许多个线程持有。典型的共享锁实现就是Semaphore(信号量)，下面就以Semaphore的公平锁模式为例子分析一下获取共享锁的源码。

    abstract static class Sync extends AbstractQueuedSynchronizer {

        private static final long serialVersionUID = 1192457210091910933L;

        //这里可以知道Semaphore的初始化的许可个数实际上就是state的值设置

        Sync(int permits) {

            setState(permits);

        }

        //获取可用许可的个数也是state的值

        final int getPermits() {

            return getState();

        }

        //暂时忽略其他代码

    }

    static final class FairSync extends Sync {

        private static final long serialVersionUID = 2014338818796000944L;

        FairSync(int permits) {

            super(permits);

        }

        /**

         * 尝试获取共享锁，如果获取成功，返回state - acquires

         */

        protected int tryAcquireShared(int acquires) {

            for (;;) {

                //当前的线程不在等待队列的头节点的直接后继节点，有其他线程占用该节点的时候返回true，此时当前线程肯定是无法获得锁的。

                if (hasQueuedPredecessors())

                    return -1;

                int available = getState();

                //计算剩余许可的值(下叫剩余值)

                int remaining = available - acquires;

                //如果剩余值小于0直接返回，或者剩余值通过CAS更新到state则返回剩余值

                if (remaining < 0 ||

                    compareAndSetState(available, remaining))

                    return remaining;

            }

        }

    }

    /**

     * 获取共享锁

     */

    public final void acquireShared(int arg) {

        // 调用需要子类覆盖的tryAcquireShared方法，方法返回值小于0则调用doAcquireShared，这里是一个快速尝试

        if (tryAcquireShared(arg) < 0)

            doAcquireShared(arg);

    }

    /**

     * 获取共享锁并且进行CLH队列的管理

     *

     */

    private void doAcquireShared(int arg) {

        //当前线程包装为Node(标记模式为SHARED)插入到CLH队列队尾

        final Node node = addWaiter(Node.SHARED);

        boolean failed = true;

        try {

            boolean interrupted = false;

            for (;;) {

                final Node p = node.predecessor();

                // 如果当前节点的前驱节点为head，调用tryAcquireShared

                if (p == head) {

                    int r = tryAcquireShared(arg);

                    if (r >= 0) {

                        //调用tryAcquireShared的返回值大于0，也就是获取共享锁成功，设置head节点

                        setHeadAndPropagate(node, r);

                        p.next = null; // help GC

                        if (interrupted)

                            selfInterrupt();

                        failed = false;

                        return;

                    }

                }

                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())

                    interrupted = true;

            }

        } finally {

            if (failed)

                cancelAcquire(node);

        }

    }

    /**

     * 设置head节点和传播唤醒状态

     *

     */

    private void setHeadAndPropagate(Node node, int propagate) {

        //把当前的head封闭在方法栈上，用以下面的条件检查，propagate值为tryAcquireShared的返回值，也就是state的"剩余值"

        Node h = head; // Record old head for check below

        //设置head节点为当前节点

        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值为tryAcquireShared的返回值，也就是state的"剩余值"，它是用于判断是否进行传播唤醒。

          * 这里的判断条件是propagate大于0，或者head节点为null，或者head节点的等待状态小于0，或者重新同步的head节点为null，

          * 或者重新同步的head节点的等待状态小于0，并且当前节点的直接后继节点不为null并且是在共享模式下才调用doReleaseShared

          */

        if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {

            Node s = node.next;

            if (s == null || s.isShared())

                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 (;;) {

            //注意，这里h就是CLH队列的头节点，在唤醒后继节点的时候，head有可能变更，也就是h跟head不相等

            Node h = head;

            //准入条件：头节点不为null(已经初始化)同时头节点不等于尾节点(也就是CLH队列中head存在后继节点(等待线程))

            if (h != null && h != tail) {

                int ws = h.waitStatus;

                //如果h节点状态为Node.SIGNAL(-1)，如果CAS设置h状态为0则对h节点的后继节点进行唤醒，否则continue

                if (ws == Node.SIGNAL) {

                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))

                        continue;            // loop to recheck cases

                    unparkSuccessor(h);

                }

                 /**

                  * 如果h节点状态为0则CAS设置h节点的状态为Node.PROPAGATE(-3)，这里没有调用unparkSuccessor，只是确保Node.PROPAGATE能够传递下去

                  * 假设只循环了一次(h节点仍然为head)，如果唤醒了后继节点，那么后继节点最终也会调用setHeadAndPropagate方法。

                  * 假设只循环了一次(h节点仍然为head)，h(头)节点等待状态更新为Node.PROPAGATE，

                  * 获取锁的线程在执行setHeadAndPropagate时可以读到PROPAGATE，从而由获取锁的线程去释放后继等待线程。

                  * 也就是说，这个方法就是为了在acquire和release竞态的情况下，保证后继节点能够被正常唤醒。

                  */

                else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))

                    continue;                // loop on failed CAS

            }

            if (h == head)                   // loop if head changed

                break;

        }

    }

释放共享锁

下面就以Semaphore的公平锁模式为例子分析一下释放共享锁的源码。

    public final boolean releaseShared(int arg) {

        //调用tryReleaseShared返回true后调用上面说到的doReleaseShared

        if (tryReleaseShared(arg)) {

            doReleaseShared();

            return true;

        }

        return false;

    }

    //这里子类覆盖的方法就是CAS设置state为state + releases成功才返回true。

    protected final boolean tryReleaseShared(int releases) {

        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;

        }

    }

这里可以看到共享锁的获取和释放都会涉及到doReleaseShared，也就是后继线程的唤醒。关于Node.PROPAGATE的作用，个人功力还不够，可以参看一下前边的参考资料里面的文章。

小结

说实话，看AQS的源码感觉比较烧脑，其中遇到不少问题需要参考不少资料或者求助stackoverflow。理解AQS的源码之后，理解其他基于AQS扩展出来的框架就相对简单了。这是第一次花了几天时间看了AQS的源码，感觉还是有点肤浅，晚点再深入看一次，希望有更多收获。

（未完待续）

巴特西

AbstractQueuedSynchronizer源码分析

AbstractQueuedSynchronizer源码分析

前提

预备知识

CLH队列

阻塞和唤醒

AQS的实现原理

1. 同步状态的原子性管理

2. 线程的阻塞与解除阻塞

3. 同步队列的管理

AQS主要数据结构

Node结构

Node状态迁移

AQS基本属性

AQS的主要方法源码解读

获取独占锁

注意点一

注意点二

释放独占锁

图解获取和释放独占锁

单线程重入

多线程抢占

获取共享锁

释放共享锁

小结

最新文章

热门文章