JAVA并发(1)-AQS(亿点细节)

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

AQS(AbstractQueuedSynchronizer), 可以说的夸张点，并发包中的几乎所有类都是基于AQS的。

一起揭开AQS的面纱

1. 介绍

为依赖 FIFO阻塞队列的阻塞锁和相关同步器 (semaphores, events等)的实现提供一个框架。为那些依赖于原子state的同步器提供基础(CyclicBarrier、CountDownLatch等). 支持独占模式和共享模式 , 不同的模式需要实现不同的方法.

引用这位大佬的图 http://www.cnblogs.com/waterystone/p/4920797.html

这个图是AQS整体结构，从图中可以看到，AQS维护着一个阻塞队列(当线程获取资源失败时，就会进入该队列，等待，直到被唤醒), state是一个 共享的资源。

2. 源码剖析

我们先看看AQS的类图，

内部类: Node，阻塞队列维护的元素；ConditionObject, 支持独占模式下的子类用作Condition实现, 后面会讲到。先看看Node的结构。

static final class Node {

	/** 表明阻塞一个共享模式的结点 */

        static final Node SHARED = new Node();

        /** 表明阻塞一个独占模式的结点 */

        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */

	// 发生的情况: 中断或者超时

        static final int CANCELLED = 1;

        /** 表明后继节点需要被唤醒 */

	// 发生的情况: 后继结点被park()阻塞，当目前的结点释放或取消，必须要unpark它的后继结点

        static final int SIGNAL = -1;

        /** waitStatus value to indicate thread is waiting on condition */

	// 发生的情况: 当前结点在条件队列中(后面会讲解)

        static final int CONDITION = -2;

        /**

         * waitStatus value to indicate the next acquireShared should

         * unconditionally propagate

         */

        // 发生的情况: 执行releaseShared后，需要传播释放其他结点。

        // 在doReleaseShared中进行设置。(后面讲解这个状态的必要性)

        // 在共享模式下，才会用到

        static final int PROPAGATE = -3;

	/** 结点的状态, 初始值为0*/

	volatile int waitStatus;

	volatile Node prev;

	volatile Node next;

	volatile Thread thread;

	 ....

}

tip: waitStatus > 0, 即 CANCELLED，此时的结点不正常。

AQS使用了模板模式, 自主选择重新定义以下方法

tryAcquire - 独占模式
tryRelease - 独占模式
tryAcquireShared - 共享模式
tryReleaseShared - 共享模式
isHeldExclusively

调用这些方法，都会引发 UnsupportedOperationException ，后面的文章将通过其子类，来讲解它们的实现。

有了这些知识后，我们从下面这几个关键的共有方法入手去讲解AQS

acquire(int arg) - 独占模式
release(int arg) - 独占模式
acquireShared(int arg) - 共享模式
releaseShared(int arg) - 共享模式

2.1 acquire

独占模式下的获取资源，忽略中断。调用tryAcquire至少一次，若成功就返回。否则，将线程入队，并可能反复阻塞和接触阻塞，并调用tryAcquire直至成功。此方法可以用于实现 Lock.lock

    public final void acquire(int arg) {

	// (2.1.1)

        if (!tryAcquire(arg) &&

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

            selfInterrupt();

    }

tryAcquire在后面文章结合子类进行分析。

代码(2.1.1) 中，此时调用了tryAcquire，获取资源失败，返回false，继续执行后续方法。

addWaiter -- Queuing utilities

使用当前线程和给定mode，新建一个Node，并且将新建Node入队

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

           // (2.1.2) 记住这个地方, 后面有个知识点会用到

            node.prev = pred;

           // (2.1.3)

            if (compareAndSetTail(pred, node)) {

                pred.next = node;

                return node;

            }

        }

        // 插入node入队，初始化如果有必要

        // 执行到这的情况:

        // 1. 阻塞队列为空时，即tail == null

        // 2. 代码(2.1.3), CAS失败

        enq(node);

        return node;

    }

enq -- Queuing utilities

一直循环直到node入队成功

    private Node enq(final Node node) {

        for (;;) {

            Node t = tail;

            // 初始时，head与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;

                }

            }

        }

    }

经过了上面的操作，目前的线程已经加入了队尾，此时做的事情就是 阻塞自己，等待资源释放并且获取，然后执行自己的操作

acquireQueued

以独占且不可中断的模式，获取已经在阻塞队列中的线程。若在等待时被中断，返回true

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

        boolean failed = true;

        try {

            boolean interrupted = false;

            // 不断自旋直到获取到资源或被park

            for (;;) {

                final Node p = node.predecessor();

                // 若node的前继结点是head，执行tryAcquire，尝试获取资源(可能刚好释放了资源，就可以不用阻塞)

                // (2.1.4)

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

                    setHead(node);

                    p.next = null; // help GC

                    failed = false;

                    return interrupted;

                }

                // 检测是否需要被park,若需要就进行park，并返回在等待时是否被中断

                if (shouldParkAfterFailedAcquire(p, node) &&

                    parkAndCheckInterrupt())

                    interrupted = true;

            }

        } finally {

            //  失败，发生的情况 1. 被中断 2. 超时.(其他方法调用该方法时，会发生)

            if (failed)

                cancelAcquire(node);

        }

    }

shouldParkAfterFailedAcquire -- Utilities for various versions of acquire

对获取资源失败的node，检测并获取结点。返回true，如果线程需要阻塞

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

        int ws = pred.waitStatus;

        // 前驱结点的状态为SIGNAL，此时就可以安心的park，等待前驱节点释放资源，然后唤醒自己

        if (ws == Node.SIGNAL)

            return true;

        // 此时ws为CANCELLED

        if (ws > 0) {

            /*

             * 前驱结点状态为CANCELLED，跳过前驱结点并重试， 直到前驱节点不为CANCELLED

             */

            do {

                node.prev = pred = pred.prev;

            } while (pred.waitStatus > 0);

            pred.next = node;

        } else {

            /*

             * waitStatus此时的值要么是0要么是PROPAGATE.

             * 需要把前驱结点状态设置为SIGNAL

             */

            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);

        }

        return false;

    }

cancelAcquire -- Utilities for various versions of acquire

取消正在进行的获取资源的尝试

    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;

        // 获取过滤后的前驱的后驱节点

        Node predNext = pred.next;

        node.waitStatus = Node.CANCELLED;

        // node为tail，就直接从队列中删除

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

            compareAndSetNext(pred, predNext, null);

        } else {

            int ws;

            // 1. 前驱结点不是头结点，该结点不是老二

            // 2. 满足以下任意一个条件：

                  // 2.1 此时的前驱结点状态为SIGNAL

                  // 2.2 此时的前驱结点状态为PROPAGATE或CONDITION，成功将状态设置为SIGNAL

            // 3. 前驱节点的任务不为空

            // 满足上面的条件，就将节点的前驱结点的next 指向 节点的后驱结点

            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 {

                // 此时结点刚好是老二;

                // 代码(2.1.4) 可以看出，头结点要么是哨兵结点，要么是已经获取到资源的结点。

                // 此时唤醒node的后驱结点，是为了防止后驱结点一直阻塞

                unparkSuccessor(node);

            }

            node.next = node; // help GC

        }

    }

unparkSuccessor

后驱结点存在一个在正在等待的结点，则唤醒它

private void unparkSuccessor(Node node) {

        int ws = node.waitStatus;

        // 将当前的结点的waitStatus设置为0，失去SIGNAL、PROPAGATE的含义

        if (ws < 0)

            compareAndSetWaitStatus(node, ws, 0);

        // (2.1.5)

        Node s = node.next;

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

            s = null;

            // [问题一] 为什么node的后驱结点为空，重新寻找是从后往前找

            // 只要waitStatus <= 0, 都有机会被唤醒

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

                // (2.1.6)

                if (t.waitStatus <= 0)

                    s = t;

        }

        if (s != null)

            LockSupport.unpark(s.thread);

    }

2.1.1 acquire流程图

根据上面的分析，整一个流程图

2.1.2 小结

根据acquire流程图，一句话小结其流程，尝试获取资源，失败则将新建node(当前线程及独占模式)入队，检测自己是否是老二，是老二就再一次尝试获取资源，成功就返回中断标志，不是老二就设置为SIGNAL，park自己，然后安心等待被唤醒。

2.2 release

独占模式 下的释放资源。解除阻塞一个或多个线程，当tryRelease返回true时。此方法可以用于实现 Lock.unlock

    public final boolean release(int arg) {

        // tryRelease返回true，继续下面操作。

        if (tryRelease(arg)) {

            Node h = head;

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

                unparkSuccessor(h);

            return true;

        }

        return false;

    }

2.2.1 release流程图

因为release(int arg)的主要流程是在 tryRelease 和 unparkSuccessor 中，但是 tryRelease 又是在子类中实现，所以该流程图也可以看作 unparkSuccessor 的流程图

2.2.2 小结

根据release流程图，一句话小结其流程，释放资源，唤醒后驱没有 被取消 的结点。

下面讲讲AQS的另一种模式， 共享模式

2.3 acquireShared

共享模式下获取资源，忽略中断。至少调用tryAcquireShared一次，成功就返回。否则，线程将入队，可能会重复的阻塞和解除阻塞，直到调用tryAcquireShared成功。成功获取到资源， 将会唤醒后驱结点 ，若资源满足。

    public final void acquireShared(int arg) {

        // 在子类探究tryAcquireShared

        if (tryAcquireShared(arg) < 0)

            doAcquireShared(arg);

    }

对tryAcquireShared返回的参数，进行简单的介绍

返回负数表示失败；
返回零，随后的线程都不能获取到资源
返回正数，随后的线程可以获取到资源

此时tryAcquireShared的返回值是小于零，表示获取资源失败，进行下一步处理

doAcquireShared

获取资源在不可中断的模式下

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

                    // 若前驱结点是头结点且刚好释放了，尝试获取资源

                    // (2.3.1)

                    if (r >= 0) {

                        // 将当前node设置为head，并且尝试唤醒node的后驱结点

                        setHeadAndPropagate(node, r);

                        p.next = null; // help GC

                        if (interrupted)

                            selfInterrupt();

                        failed = false;

                        return;

                    }

                }

                // 跟独占模式的处理是一样的

                // (2.3.2)

                if (shouldParkAfterFailedAcquire(p, node) &&

                    parkAndCheckInterrupt())

                    interrupted = true;

            }

        } finally {

            if (failed)

                cancelAcquire(node);

        }

    }

setHeadAndPropagate

设置队列的头结点，达到条件就唤醒后面的结点.

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.

         */

        // tips: h == null、(h = head) == null、s == null，是为了防止空指针的发生

        // 执行到下面的情况:

        // 1. propagate是tryAcquireShared返回的值。propagate(资源) > 0， 表示还有资源可以唤醒后面的结点。

        // 否则，此时propagate = 0, 结合代码(2.3.1)

        // 2. 旧的head的waitStatus < 0

        // 旧的头结点释放了资源，执行了代码(2.3.4), 此时的waitStatus 为PROPAGATE(初始化为0)

        // 3. 此时的head已经是当前结点了，后面若有结点(此时后面的结点在park)，

        // 那么新head的waitStatus肯定为SIGNAL, 结合代码(2.3.2)

        // 情况3，有可能会发生没必要的唤醒，因为此时去唤醒新head的后驱结点，但是此时还没

        // 有释放资源，它后驱结点唤醒后，去获取资源，获取失败，又被park.

        // 源码注释说到，虽然有争议，但是大多数情况下，需要去唤醒

        // (2.3.3)

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

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

            Node s = node.next;

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

                doReleaseShared();

        }

    }

代码(2.3.3)中, 为什么不只用propagate来判断是否唤醒后驱结点 [问题二]

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;

                if (ws == Node.SIGNAL) {

                    // (2.3.4)

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

                        continue;            // loop to recheck cases

                    unparkSuccessor(h);

                }

                // (2.3.5)

                else if (ws == 0 &&

                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))

                    continue;                // loop on failed CAS

            }

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

                break;

        }

    }

2.3.1 acquireShared的流程图

2.3.2 小结

一句话小结acquireShared的流程，尝试获取资源，若获取到资源，资源还有剩余就去继续唤醒后驱结点，若尝试获取资源失败，就park自己，等待被唤醒。跟 acquire 相比，最大的区别就是，获取到资源acquireShared， 还会去尝试唤醒其后驱结点

2.4 releaseShared

Releases in shared mode. Implemented by unblocking one or more threads if tryReleaseShared returns true.

    // 代码比较简单，就不分析了~

    public final boolean releaseShared(int arg) {

        if (tryReleaseShared(arg)) {

            doReleaseShared();

            return true;

        }

        return false;

    }

2.5 问题解答

[问题一] 为什么在唤醒后驱结点时，node的后驱结点为空，需要重新从后往前找

   // (2.1.2) 记住这个地方, 后面有个知识点会用到

   node.prev = pred; // 1

   // (2.1.3)

   if (compareAndSetTail(pred, node)) { // 2

      pred.next = node; // 3

      return node;

   }

仔细观察代码(2.1.2) 和(2.1.3), 此时添加结点相当于有三步，都不是原子性的，当执行到第二步时，就要唤醒后驱结点了，此时新增的结点只设置了 前驱结点 ，队列设置了尾结点，但是 没有设置后驱结点 ，如果从前往后查找的话，可能会丢失符合要求的结点。

[问题二] ，代码(2.3.3)中, 为什么不只用propagate来判断是否唤醒后驱结点。

请看这位大佬的博客讲的非常详细

大致意思就是,

我们假设有A、B、C、D四个线程，前两个释放资源的线程，后两个是争抢资源的线程，此时只有A或B释放了资源，C、D才可以被唤醒， 假设我们不看PROPAGTE

时刻一: A线程释放资源，执行 代码 (2.3.4) ，head的waitStatus从SIGNAL(-1)变为了0

时刻二: C线程获取到资源，执行到代码 (2.3.1) , tryAcquireShared返回0

时刻三: B线程线程释放资源，执行 代码 (2.3.4) ，因为此时未改变头结点，head的waitStatus为0，不能unparkSuccessor

时刻四: 此时C执行到 代码(2.3.3) ，propagate(tryAcquireShared返回值)为0，C也不会去唤醒后驱结点，D线程就永远GG了

引用 doReleaseShared 注释中的一句话

status is set to PROPAGATE to ensure that upon release, propagation continues.

2.6 Condition

使用synchronized时，线程间通信使用 wait, notify and notifyAll ;而使用AQS实现的lock，线程间的通信就使用Condition中的 await、signal... 。Condition与Lock结合使用，同一个lock对应多个Condition。

public class ConditionObject implements Condition...

在AQS中，已经对Condition的方法进行了实现，子类想使用的话，只需要调用ConditionObject就行了

        final ConditionObject newCondition() {

            return new ConditionObject();

        }

本来想跟着源码走，简简单单介绍一下Condition，但是源码有几处细节，让我头秃，在网上搜索别人的博客，这篇博客解开了我的疑惑，对Condition介绍的非常详细，写的非常的完美~

根据大佬的博客，那我们下面简单讲解Condition的两个常用方法

await
signal

2.6.1 await & signal

导致目前线程阻塞直到被唤醒或中断 ;调用await后，会将当前的线程封装成node，加入到 条件队列 中

        public final void await() throws InterruptedException {

            if (Thread.interrupted())

                throw new InterruptedException();

            // 添加结点到条件队列中

            Node node = addConditionWaiter();

            // 释放当前线程持有的锁

            int savedState = fullyRelease(node);

            int interruptMode = 0;

            // 不在同步队列中，就park

            // (2.6.1)

            while (!isOnSyncQueue(node)) {

                // 执行到这，当前线程会立即挂起

                LockSupport.park(this);

                // 运行到这的话，情况: 1. signal 2. 中断

                // 检验中断

                // (2.6.2)

                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

                    break;

            }

            // 第二部分

            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

                interruptMode = REINTERRUPT;

            if (node.nextWaiter != null) // clean up if cancelled

                unlinkCancelledWaiters();

            if (interruptMode != 0)

                reportInterruptAfterWait(interruptMode);

        }

为了讲清楚 代码(2.6.1) 之后的逻辑，我们先看看signal的源码

将condition queue中 等待最长 的结点转移到sync queue中去，去争抢资源

        public final void signal() {

            if (!isHeldExclusively())

                throw new IllegalMonitorStateException();

            Node first = firstWaiter;

            if (first != null)

                doSignal(first);

        }

此时，执行signal的主要逻辑

        private void doSignal(Node first) {

            do {

                if ( (firstWaiter = first.nextWaiter) == null)

                    lastWaiter = null;

                // 将后驱结点置空

                // (2.6.3)

                first.nextWaiter = null;

            } while (!transferForSignal(first) &&

                     (first = firstWaiter) != null);

        }

将condition queue中的一个结点转移到sync queue中去

    final boolean transferForSignal(Node node) {

        // 这里表示当前已经被取消了。

        // (2.6.4)

        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))

            return false;

        // 将当前结点放入sync queue的末尾， 此时返回的是当前结点的前驱结点(一定要注意)

        Node p = enq(node);

        int ws = p.waitStatus;

        // 前驱结点被取消，或者设置SIGNAL状态失败，就直接唤醒当前线程, 唤醒 = 有资格去竞争锁了

        // enq返回的是前驱结点，我人傻了，看成是返回当前线程，就一直觉得逻辑不对。

        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))

            LockSupport.unpark(node.thread);

        return true;

    }

这里signal的逻辑就讲完了,总结一下:

在 condition queue 中找出等待时间最长且未被取消的结点，转移到 sync queue 的队尾去，同时要在condition queue中删除该结点
若在sync queue中的该结点的前驱结点被取消了或设置SIGNAL状态失败，要直接唤醒它，叫它去竞争锁。

signalAll 的主要逻辑和 signal 是一样的，差别就是 signalAll 会把所有在condition queue中的结点转移到sync queue中去，并清空所有在condition queue中的结点，下面只贴一下 signalAll 的主要代码，

        private void doSignalAll(Node first) {

            lastWaiter = firstWaiter = null;

            do {

                Node next = first.nextWaiter;

                first.nextWaiter = null;

                transferForSignal(first);

                first = next;

            } while (first != null);

        }

我们再次回到await中去

        public final void await() throws InterruptedException {

            if (Thread.interrupted())

                throw new InterruptedException();

            // 添加结点到条件队列中

            Node node = addConditionWaiter();

            // 释放当前线程持有的锁

            int savedState = fullyRelease(node);

            int interruptMode = 0;

            // 不在同步队列中，就park

            // (2.6.1)

            while (!isOnSyncQueue(node)) {

                // 执行到这，当前线程会立即挂起

                LockSupport.park(this);

                // 运行到这的话，情况: 1. signal 2. 中断

                // 检验中断

                // (2.6.2)

                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

                    break;

            }

            // 第二部分

            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

                interruptMode = REINTERRUPT;

            if (node.nextWaiter != null) // clean up if cancelled

                unlinkCancelledWaiters();

            if (interruptMode != 0)

                reportInterruptAfterWait(interruptMode);

        }

从 代码(2.6.1) ，继续讲解

isOnSyncQueue

一开始在条件队列中，现在在sync queue中等待重新获取资源，如果有这种的node就返回true

    final boolean isOnSyncQueue(Node node) {

        if (node.waitStatus == Node.CONDITION || node.prev == null)

            return false;

        if (node.next != null) // If has successor, it must be on queue

            return true;

        /*

         * node.prev can be non-null, but not yet on queue because

         * the CAS to place it on queue can fail. So we have to

         * traverse from tail to make sure it actually made it.  It

         * will always be near the tail in calls to this method, and

         * unless the CAS failed (which is unlikely), it will be

         * there, so we hardly ever traverse much.

         */

        return findNodeFromTail(node);

    }

findNodeFromTail

从尾部找寻结点

    private boolean findNodeFromTail(Node node) {

        Node t = tail;

        for (;;) {

            // 结合下面的enq代码以及图思考一下，会明白此方法的意义的

            if (t == node)

                return true;

            if (t == null)

                return false;

            t = t.prev;

        }

    }

node.waitStatus == Node.CONDITION , 表示当前结点肯定在 condition queue 中。

为何是上面的那些条件?

我们上面看了转移到 sync queue 是用的 enq 方法

    private Node enq(final Node node) {

        for (;;) {

            Node t = tail;

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

                if (compareAndSetHead(new Node()))

                    tail = head;

            } else {

                // (2.6.5)

                node.prev = t;

                if (compareAndSetTail(t, node)) {

                    t.next = node;

                    return t;

                }

            }

        }

    }

结合 代码(2.6.5) ,思考一下就知道 isOnSyncQueue 中条件设置的道理了，但是为何需要 findNodeFromTail 啦？这是需要补充一个知识点了，在多个线程执行 enq 时，只有一个线程会设置为 tail 成功，其余的都只是设置 prev ，就可能会出现下面图片中的情况， '多个尾巴' 。 一直不断自旋 ，最后会变成一个正常的链表。

此时线程的状态是，调用await后，将结点添加到条件队列中，且释放了自己持有的 所有资源 ，并将自己park，此时等待被signal或者被中断。

        public final void await() throws InterruptedException {

            if (Thread.interrupted())

                throw new InterruptedException();

            // 添加结点到条件队列中

            Node node = addConditionWaiter();

            // 释放当前线程持有的锁

            int savedState = fullyRelease(node);

            int interruptMode = 0;

            // 不在同步队列中，就park

            // (2.6.1)

            while (!isOnSyncQueue(node)) {

                // 执行到这，当前线程会立即挂起

                LockSupport.park(this);             ///  我们在这被挂起了

                // 运行到这的话，情况: 1. signal 2. 中断

                // 检验中断

                // (2.6.2)

                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

                    break;

            }

            // 第二部分

            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

                interruptMode = REINTERRUPT;

            if (node.nextWaiter != null) // clean up if cancelled

                unlinkCancelledWaiters();

            if (interruptMode != 0)

                reportInterruptAfterWait(interruptMode);

        }

执行到 代码(2.6.2) , 我们直到可能是被signal或被中断了。现在要解决的是，

是否被中断?
何时被中断?
中断如何处理?

我们带着这三个问题，继续出发~

补充一个小知识点，AQS定义了三种情况中断的值

THROW_IE, signal前被中断，要抛出InterruptedException
REINTERRUPT， signal后被中断
0，未被中断

关于 REINTERRUPT 这个中断，可以理解成，吃饭，吃完了但是还有一个菜没有上，问服务员，“如果没有炒，就不要了”，但是服务员告诉，菜已经下锅，所以这时候的中断就是 REINTERRUPT ，中断的太晚了。 -- 例子来自上面的那篇博客

我们继续看向 代码(2.6.2)

checkInterruptWhileWaiting

        private int checkInterruptWhileWaiting(Node node) {

            return Thread.interrupted() ?

                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :

                0;

        }

若被中断 Thread.interrupted 肯定为 true

transferAfterCancelledWait

    final boolean transferAfterCancelledWait(Node node) {

        // 中断情况一: 此时结点还在condition queue中，肯定是signal前就被中断了

        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {

            // 这里还要加入到sync queue中，获取到锁才能抛出错，继续往后看

            enq(node);

            return true;

        }

        /*

         * If we lost out to a signal(), then we can't proceed

         * until it finishes its enq().  Cancelling during an

         * incomplete transfer is both rare and transient, so just

         * spin.

         */

         // 中断情况二: 这里的情况就是signal后，node还没有执行enq，毕竟执行signal到执行enq还有几个步骤

         // 此时就自旋，等待node转移到sync node中就行了

        while (!isOnSyncQueue(node))

            Thread.yield();

        return false;

    }

上面的代码已经注明了，各种情况的发生时机，此时我们来到了await的第二部分~

        public final void await() throws InterruptedException {

            if (Thread.interrupted())

                throw new InterruptedException();

            // 添加结点到条件队列中

            Node node = addConditionWaiter();

            // 释放当前线程持有的锁

            int savedState = fullyRelease(node);

            int interruptMode = 0;

            // 不在同步队列中，就park

            // (2.6.1)

            while (!isOnSyncQueue(node)) {

                // 执行到这，当前线程会立即挂起

                LockSupport.park(this);             ///  我们在这被挂起了

                // 运行到这的话，情况: 1. signal 2. 中断

                // 检验中断

                // (2.6.2)

                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

                    break;

            }

            // 第二部分

            // acquireQueued获取到锁，并返回是否被中断。

            // 有一种情况需要提一下，acquireQueued返回true，上面的interruptMode = 0，

            // 表示signal后，在获取锁的时候被中断了

            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

                interruptMode = REINTERRUPT;

            // node还在condition上，说明是被取消了的node，清除它

            if (node.nextWaiter != null) // clean up if cancelled

                unlinkCancelledWaiters();

            if (interruptMode != 0)

                // 对上面得到的interruptMode做出回应

                reportInterruptAfterWait(interruptMode);

        }

reportInterruptAfterWait

        private void reportInterruptAfterWait(int interruptMode)

            throws InterruptedException {

            if (interruptMode == THROW_IE)

                throw new InterruptedException();

            else if (interruptMode == REINTERRUPT)

                selfInterrupt();

        }

2.6.2 await与signal的流程图

2.6.3 小结

先讲讲 await的流程 ，看起流程图有点吓人，其实很多步骤是 对不同时机的中断操作的记录

当await被执行，下面简单总结下await的流程

将 当前线程与CONDITION状态 封装成 node ，加入到 condition queue 的末尾
释放线程之前获取的 所有资源
若不在 sync queue 中，阻塞自己，等待被 signal 或 被中断
获取中断操作的时机，并记录表示何时中断的值( interruptMode )
不管是怎么被唤醒的，都要 去竞争资源 ，直到获得资源为止
最后对不同的中断值，作出不同的操作

signal的流程 就相对于简单一点

获取 condition queue 的头结点
检验是否被取消，若被取消，就获取头结点的后驱结点，以此类推；
将结点从 condition queue 中转移到 sync queue 中，而且会从 condition queue 中删除该节点
若结点插入 sync queue ，得到的前驱结点，被取消了，或者CAS前驱结点状态为 SIGNAL

失败，将直接unpark当前线程

3. 总结

Doug Lea ，太秀了。AQS中有很多细枝末节的东西，只有自己去读了源码，理解为何这样做，你才会明白才会真正读懂AQS。

关于学习和写AQS文章时，看了一些博客，为我解答了自己的疑惑，慢慢加油，我也要向这些大佬看齐~

4. 参考

《Java并发编程之美》 - 这本书可以作为学习并发的入门书

Java并发之AQS详解 - 引用了他的图片

AbstractQueuedSynchronizer源码解读 - 为我解开了一些获取和释放资源的疑惑

逐行分析AQS源码(4)——Condition接口实现 - 为我解开了一些 Condition 的疑惑

AQS论文 Doug Lea

~~# 5. 面试中问题

这是我的一个想法，若我博客中写过的知识，在面试中有问到过，我会记录下来，没有就是目前还没遇到过

正文

1. 介绍

为依赖 FIFO阻塞队列的阻塞锁和相关同步器 (semaphores, events等)的实现提供一个框架。为那些依赖于原子state的同步器提供基础(CyclicBarrier、CountDownLatch等). 支持独占模式和共享模式 , 不同的模式需要实现不同的方法.

2. 源码剖析

2.1.1 acquire流程图

2.1.2 小结

2.2.1 release流程图

2.2.2 小结

2.3.1 acquireShared的流程图

2.3.2 小结

2.4 releaseShared

Releases in shared mode. Implemented by unblocking one or more threads if tryReleaseShared returns true.

`// 代码比较简单，就不分析了~ public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }`

2.5 问题解答

2.6 Condition

2.6.1 await & signal

2.6.2 await与signal的流程图

2.6.3 小结

3. 总结

相关阅读

ios系统的动画细节，iphone的动画效果在哪里设置

macos系统细节，apple mac 系统

代码架构设计细节，代码构建是什么过程

go语言更新细节，golang 更新

项目资料管理注意细节，如何做好项目资料管理工作

股权架构设计落地细节，股权架构设计方案与组织架构

项目实施细节管理，项目管理细化

mysql聚簇索引细节，sql 聚簇索引

目录[+]

1. 介绍 为依赖 FIFO阻塞队列 的 阻塞锁 和 相关同步器 (semaphores, events等)的实现提供 一个框架 。 为那些依赖于原子state的同步器提供基础(CyclicBarrier、CountDownLatch等). 支持独占模式 和 共享模式 , 不同的模式需要实现不同的方法.

2. 源码剖析

2.1.1 acquire流程图

2.1.2 小结

2.2.1 release流程图

2.2.2 小结

2.3.1 acquireShared的流程图

2.3.2 小结

2.4 releaseShared Releases in shared mode. Implemented by unblocking one or more threads if tryReleaseShared returns true. // 代码比较简单，就不分析了~ public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }

2.5 问题解答

2.6 Condition

2.6.1 await & signal

2.6.2 await与signal的流程图

2.6.3 小结

3. 总结

相关阅读

ios系统的动画细节，iphone的动画效果在哪里设置

macos系统细节，apple mac 系统

代码架构设计细节，代码构建是什么过程

go语言更新细节，golang 更新

项目资料管理注意细节，如何做好项目资料管理工作

股权架构设计落地细节，股权架构设计方案与组织架构

项目实施细节管理，项目管理细化

mysql聚簇索引细节，sql 聚簇索引

目录[+]

1. 介绍

为依赖 FIFO阻塞队列的阻塞锁和相关同步器 (semaphores, events等)的实现提供一个框架。为那些依赖于原子state的同步器提供基础(CyclicBarrier、CountDownLatch等). 支持独占模式和共享模式 , 不同的模式需要实现不同的方法.

2.4 releaseShared

Releases in shared mode. Implemented by unblocking one or more threads if tryReleaseShared returns true.

`// 代码比较简单，就不分析了~ public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }`