AQS(AbstractQueuedSynchronizer)的实现原理

前言

前一篇文章讲了一下AQS是什么以及AQS可重写的方法、提供的模板方法，本篇就从以下几点来写一下同步器的实现原理。

同步队列
独占式同步状态获取与释放
共享式式同步状态获取与释放
超时获取同步状态

同步队列实现原理

同步器依赖内部的同步队列(一个FIFO双向队列)来完成同步状态的管理，当前线程获取同步状态失败时，同步器会将当前线程以及等待状态等信息构建为一个节点(Node)并将其加入同步队列，同步会阻塞当前线程，当同步状态释放时，会将首节点中的线程唤醒，使其再次尝试获取同步状态。

节点的主要属性有以下几种：

属性类型与名称	描述
int waitStatus	等待状态(如CANCELLED、SIGNAL、CONDITION、PROPAGATE、INITIAL)
Node prev	前驱节点(当节点加入同步队列时被设置，在尾部添加)
Node next	后继节点
Thread thread	当前获取同步状态的线程

节点是构成同步队列的基础，同步器拥有首节点和尾节点，如果没有成功获取同步状态的线程将会成为节点添加入队列的尾部，如下图所示：如上图，同步器包含了两个节点，一个指向头节点，一个指向尾节点，当一个线程成功获取到锁，而其他线程就要被加入队列尾部，为了保证这个过程线程安全，同步器提供了一个基于CAS的设置尾节点的方法，CAS(比较再交换，比如同步器现在保存的尾节点为1，那么下一个没有获取到锁的线程要进入同步对立，就先判断队列尾部对应的是不是1，如果是1则设置成功，如果不是1则设置失败，再次循环，直到设置成功或者线程中断)，只有设置成功后，当前节点才正式与之前的尾节点建立关联。
同步器将节点加入到同步队列的过程如下图所示：

同步队列遵循先进先出规则(FIFO)，首节点是获取同步状态成功的节点，首节点的线程在释放同步状态时，会将后继节点唤醒，而后继节点将会在获取同步状态时将自己设置为首节点，如下图所示：如上图，设置首节点是通过获取同步状态成功的线程来完成的，由于只有一个线程能成功获取到同步状态，因此设置头节点的方法并不需要使用CAS来保证线程安全，它只需要将首节点设置成为原首节点的后继节点并断开原首节点的next引用即可。

独占式同步状态获取与释放实现原理

看过上一章的应该都知道，通过同步器的acquire(int arg)方法可以尝试获取同步状态，该方法对中断不敏感，就是说由于线程获取同步状态失败之后进入同步队列之后，后续对线程进行中断操作时，线程不会从同步队列中移除，代码如下：

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

上述代码主要完成了同步状态获取、节点构造、加入同步队列以及在同步队列中自旋等待的相关工作，首先调用自定义同步器实现的tryAcquire(int arg)方法，该方法保证线程安全的获取同步状态，如果获取失败，则构造同步节点并通过addWaiter(Node node)方法将该节点加入同步队列的尾部，最后调用acquireQueued(Node node, int arg)方法，使得该节点以“死循环(for(;;))”的方式获取同步状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出队或线程被中断来实现。
将节点添加到同步队列尾部的代码如下：

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

上述代码通过使用CAS方法来确保节点能够被线程安全添加。在enq(final Node node)方法中，同步器通过“死循环”来保证节点正确被添加，在“死循环”中只有通过CAS将节点设置为尾节点之后，当前线程才能从该方法中返回，否则，当前线程不断通过CAS尝试设置。
当节点进入同步队列中之后，就进入了一个自旋的过程，每个节点(就是每个线程)都在自省观察，当条件满足，获取到了同步状态，就可以从这个自旋中退出，否则依旧留在自旋过程中(并会阻塞当前节点的线程)，代码如下：

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

在acquireQueued(final Node node, int arg)方法中，当前线程在死循环中尝试获取同步状态，而之后前驱节点是头节点的线程才能尝试获取同步状态，原因如下：

头节点是成功获取到同步状态的节点，而头节点的线程释放了同步状态之后，将会唤醒其后继节点，后继节点的线程被唤醒后需要检查自己的前驱节点是否为头节点。
维护同步队列的先进先出原则。由于非首节点线程前驱节点出队或者线程被中断从等待状态返回，随后检查自己的前驱是否为首节点，如果是则尝试获取同步状态，如果不是则继续自旋。可以看到节点与节点之间基本是不通信，而是简单的判断自己的前驱是否为头节点，这样就使得节点释放规则符合先进先出，并且也便于对过早通知的处理。
acquire(int arg)方法的流程图如下：如上流程图，前驱节点为头节点且能够获取同步状态的判断条件和线程进入等待状态是获取同步状态的自旋过程。当同步状态获取成功之后，当前线程从acquire(int arg)方法返回。
当线程获取了锁，执行相应逻辑之后就要释放锁，使得后续节点能够继续获取同步状态。通过调用同步器的release(int arg)方法可以释放同步状态，该方法在释放了同步状态之后，会唤醒后继节点(进而使后继节点重新尝试获取同步状态)。代码如下：

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

其中unparkSuccessor(h)这一句就是用来唤醒处于等待状态的后继节点。

共享式式同步状态获取与释放实现原理

共享式获取与独占式获取同步状态最主要的区别在于同一时刻能否有多个线程同时获取到同步状态。以文件的读写为例，如果一个程序在对文件进行读操作，那么这一时刻对于该文件的写操作均被阻塞，而读操作能够同时进行。写操作要求对资源的独占式访问，而读操作可以是共享式访问，两种不同的访问模式在同一时刻对文件或资源的访问情况如下图所示：如上图所示，左边共享式访问资源时，其他共享式的访问均被允许，而独占式访问被阻塞，而右边独占式访问资源时，同一时刻的其他访问均被阻塞。
通过同步器的acquireShared(int arg)方法可以共享式的获取同步状态，代码如下：

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 GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

在acquireShared(int arg)方法中，同步器调用tryAcquireShared(int arg)方法尝试获取同步状态，当返回值大于0时，表示能够获取到同步状态。因此，在共享式获取的自旋过程中，成功获取到同步状态并退出自旋的条件就是tryAcquireShared(int arg)方法返回值大于等于0。可以看到，在doAcquireShared(int arg)方法的自旋过程中，如果当前节点的前驱为头节点时，尝试获取同步状态，如果返回值大于等于0，表示该次获取同步状态成功并从自旋过程中退出。
与独占式一样，共享式也需要释放同步状态，通过调用releaseShared(int arg)方法可以释放同步状态，该方法代码如下：

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

该方法在释放同步状态之后，会唤醒后续处于等待状态的节点。对于能支持多个线程同时访问的并发组件，它和独占式主要区别在于tryReleaseShared(int arg)方法必须确保同步状态线程安全释放(想一下就明白了，独占式是一个线程去释放锁，而共享式是多个线程去释放锁，所以要保证每一个线程都释放锁成功)，一般是通过循环和CAS来保证的。

超时获取同步状态实现原理

通过调用同步器的tryAcquireSharedNanos(int arg, long nanosTimeout)方法可以超时获取同步状态，即在指定的时间段内获取同步状态，如果过去到则返回true，否则返回false。该方法提供了传统Java同步操作(比如synchronized关键字)不具备的特性。
超时获取同步状态过程可以被视为相应中断获取同步状态过程的“增强版”，tryAcquireSharedNanos(int arg, long nanosTimeout)方法在支持相应中断的基础上，增强了超时获取的特性。针对超时通知获取，主要需要计算出需要睡眠的时间间隔nanosTimeout，为了防止过早通知，nanosTimeout的计算公式为：nanosTimeout = now - lastTime,其中now为当前唤醒时间，lastTime为上次唤醒时间，如果nanosTimeout大于0则表示超时时间未到，需要继续睡眠nanosTimeout纳秒，反之，表示已经超时，即返回false，代码如下：

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 GC
                        failed = 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);
        }
    }

该方法在自旋过程中，当节点的前驱节点为头节点时尝试获取同步状态，如果获取成功则从该方法返回，否则判断是否超时(nanosTimeout小于等于0表示超时)，如果没有超时，重新计算超时间隔nanosTimeout，然后使当前线程等待nanosTimeout纳秒(当已经到设置的超时时间，该线程会从LockSupport.parkNanos(this, nanosTimeout)这一行代码进行返回)。
如果nanosTimeout小于等于spinForTimeoutThreshold(1000纳秒)时，将不会使该线程进行超时等待，而是进入快速的自旋过程。原因在于：非常短的超时等待无法做到十分精确，如果这时再进行超时等待，将会让nanosTimeout的超时从整体上表现得反而不精确。因此，在超时非常短的场景下，同步器会进入无条件的快速自旋。独占式超时获取同步状态的流程图如下所示：从上图可以看出，独占式超时获取同步状态tryAcquireSharedNanos(int arg, long nanosTimeout)方法和独占式获取同步状态acquire(int arg)方法非常相似，其主要区别在于未获取到同步状态时的处理逻辑。acquire(int arg)在未获取到同步状态时，会将线程一直处于等待状态。而tryAcquireSharedNanos(int arg, long nanosTimeout)方法会使当前线程等待nanosTimeout纳秒，如果在这段时间内还没有获取到同步状态，将会从等待逻辑中自动返回。

结尾

如果看了前一篇文章，就应该对AQS不陌生了，有很多锁基于AQS来实现的，比如读写锁、计数器、可重入锁等，看懂了AQS之后再去看其他锁就会简单的，四个字概括：大道本源。你连源头都掌握了，还怕掌握不了其他的吗？
最后，如果大家有需要的话可以关注一下我的公众号，会即时更新Java相关技术文章，公众号内还有一些实用资料，如Java秒杀系统视频教程、黑马2019的教学资料(IDEA版)、BAT面试题汇总(分类齐全)、MAC电脑常用安装包(有一些是淘宝买的，已PJ的)。