聊聊高并发（二十二）解析java.util.concurrent各个组件（四） 深入理解AQS（二）

上一篇介绍了AQS的基本设计思路以及两个内部类Node和ConditionObject的实现
聊聊高并发（二十一）解析java.util.concurrent各个组件（三） 深入理解AQS（一） 这篇说一说AQS的主要方法的实现。AQS和CLHLock的最大区别是，CLHLock是自旋锁，而AQS使用Unsafe的park操作让线程进入等待（阻塞）。

线程加入同步队列，和CLHLock一样，从队尾入队列，使用CAS+轮询的方式实现无锁化。入队列后设置节点的prev和next引用，形成双向链表的结构

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

线程指定独享还是共享方式加入队列，先尝试加入一次，如果失败再用enq()轮询地尝试，比如addWaiter(Node.EXCLUSIVE), addWaiter(Node.SHARED)

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

唤醒后继节点，最典型的情况就是在线程释放锁后，会唤醒后继节点。会从节点的next开始，找到一个后继节点，如果next是null，就从队尾开始往head找，直到找到最靠近当前节点的后续节点。 waitStatus <= 0的隐含意思是线程没有被取消。 然后用LockSupport唤醒这个找到的后继节点的线程。

这个方法类似于CLHLock里面释放锁时，通知后续节点来获取锁。AQS使用了阻塞的方式，所以这个方法的后续方法是acquireXXX方法，它负责将后续节点唤醒，后续节点再根据状态去判断是否获得锁

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.
         */
        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.
         */
        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;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

共享模式下的释放操作，从队首开始向队尾扩散，如果节点的waitStatu是SIGNAL,就唤醒后继节点，如果waitStatus是0，就设置标记成PROPAGATE

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
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

取消获取操作，要把节点从同步队列中去除，通过链表操作将它的前置节点的next指向它的后继节点集合。如果该节点是在队尾，直接删除即可，否则要通知后继节点去获取锁

private void cancelAcquire(Node node) {
        // Ignore if node doesn't exist
        if (node == null)
            return;

        node.thread = null;

        // Skip cancelled predecessors
        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.
        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.
        node.waitStatus = Node.CANCELLED;

        // If we are the tail, remove ourselves.
        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.
            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 {
                unparkSuccessor(node);
            }

            node.next = node; // help GC
        }
    }

独占模式并且不可中断地获取队列锁的操作，这个方法在ConditionObject.await()中被使用，当线程被Unsafe.unpark唤醒后，需要调用acquireQueued来获取锁，从而结束await(). accquireQueued()方法要么获得锁，要么被tryAcquire方法抛出的异常打断，如果抛出异常，最后在finally里面取消获取

值得注意的是只有节点的前驱节点是head的时候，才能获得锁。这里隐含了一个意思，就是head指向当前获得锁的节点。当程序进入if(p == head and tryAcquire(arg))这个分支时，表示线程获得了锁或者被中断，将自己设置为head，将next设置为null.

shouldParkAfterFailedAcquired()方法的目的是将节点的前驱节点的waitStatus设置为SIGNAL，表示会通知后续节点，这样后续节点才能放心去park，而不用担心被丢失唤醒的通知。

parkAndCheckInterupt()方法会真正执行阻塞，并返回中断状态，这个方法有两种情况返回，一种是park被unpark唤醒，这时候中断状态为false。另一种情况是park被中断了，由于这个accquireQueued方法是不可中断的版本，所以即使线程被中断了，也只是设置了中断标志为true,没有跑出中断异常。在支持中断的获取版本里，这时会抛出中断异常。

这个方法可以理解为Lock的lock里没有获取锁的分支，在CLHLock自旋锁的实现里，是对前驱节点的状态自旋，而AQS是阻塞，所以这里是在同步队列里面进入了阻塞状态，等待被前驱节点释放锁时唤醒。

释放锁时会根据状态调用unparkSuccessor()方法来唤醒后续节点，这样就会在这个方法里面把阻塞的线程唤醒并获得锁。

队列锁的好处是线程都在多个共享状态上自旋或阻塞，所以unparkSuccessor()方法只会唤醒它后继没有取消的节点。

而取消只有两种情况，中断或者超时

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

独占模式支持中断的获取队列锁操作，可以看到和不支持中断版本的区别，这里如果parkAndCheckInterrupt()方法返回时显示被中断了，就抛出中断异常

private void doAcquireInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

独占模式限时获取队列锁操作, 这个获取的整体逻辑和前面的类似，区别是它支持限时操作，如果等待时间大于spinForTimeoutThreshold，就使用阻塞的方式等待，否则用自旋等待。使用了LockSupport.parkNanos（）方法来实现限时地等待，并支持中断

这里隐含的一个含义是parkNanos方法退出有3种方式，

1. 限时到了自动退出，这时候会超时

2. 没有到限时被唤醒了，这时候是不超时的

3. 被中断

private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
        long lastTime = System.nanoTime();
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                if (nanosTimeout <= 0)
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                long now = System.nanoTime();
                nanosTimeout -= now - lastTime;
                lastTime = now;
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

共享模式获得队列锁操作，获得操作也是从head的下一个节点开始，和独占模式只unparkSuccessor一个节点不同，共享模式下，等head的后续节点被唤醒了，它要扩散这种共享的获取，使用setHeadAndPropagate操作，把自己设置为head，并且把释放的状态往下传递，这里采用了链式唤醒的方法，1个节点负责唤醒1个后续节点，直到不能唤醒。当后继节点是共享模式isShared，就调用doReleaseShared来唤醒后继节点

doReleaseShared会从head开始往后检查状态，如果节点是SIGNAL状态，就唤醒它的后继节点。如果是0就标记为PROPAGATE, 等它释放锁的时候会再次唤醒后继节点。

这里有个隐含的意思:

1. 加入同步队列并阻塞的节点，它的前驱节点只会是SIGNAL，表示前驱节点释放锁时，后继节点会被唤醒。shouldParkAfterFailedAcquire()方法保证了这点，如果前驱节点不是SIGNAL,它会把它修改成SIGNAL。这里不是SIGNAL就有可能是PROPAGATE

2. 造成前驱节点是PROPAGATE的情况是前驱节点获得锁时，会唤醒一次后继节点，但这时候后继节点还没有加入到同步队列，所以暂时把节点状态设置为PROPAGATE,当后继节点加入同步队列后，会把PROPAGATE设置为SIGNAL,这样前驱节点释放锁时会再次doReleaseShared，这时候它的状态已经是SIGNAL了，就可以唤醒后续节点了

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

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) 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) {
            Node s = node.next;
            if (s == null || s.isShared())
                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))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

tryXXXX 方法，这几个方法是给子类重写的，用来扩展响应的同步器操作

protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }

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

protected boolean tryReleaseShared(int arg) {
        throw new UnsupportedOperationException();
    }

独占模式获取操作的顶层方法，如果没有tryAcquired，或者没有获得队列锁，就中断

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

独占模式释放操作的顶层方法，如果tryRelease()成功，那么就唤醒后继节点去获取锁

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