并发编程 AQS 思想之AbstractQueuedSynchronizer类的源码解析（十七）

我不是女神ヾ 2022-12-28 00:51 197阅读 0赞

今天我们分享并发编程 AQS 思想之AbstractQueuedSynchronizer类的源码解析：

***一、AQS***  ***中的数据结构*** ***-*** ***节点和同步队列***

1、节点  Node 类

既然说  Java  中的  AQS  是  CLH  队列锁的一种变体实现，毫无疑问，作为队列来 说，必然要有一个节点的数据结构来保存我们前面所说的各种域，比如前驱节点， 节点的状态等，这个数据结构就是 AQS  中的内部类  Node 。作为这个数据结构应 该关心些什么信息？

1 、线程信息，肯定要知道我是哪个线程；

2 、队列中线程状态，既然知道是哪一个线程，肯定还要知道线程当前处在 什么状态，是已经取消了“获锁”请求，还是在“”等待中”，或者说“即将得 到锁”

3 、前驱和后继线程，因为是一个等待队列，那么也就需要知道当前线程前 面的是哪个线程，当前线程后面的是哪个线程（因为当前线程释放锁以后，理当 立马通知后继线程去获锁）。

所以这个  Node 静态内部 类是这么设计的：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

4、源码参数含义如下：

static final class Node {
        /**  
         * 表示线程以共享的模式等待锁（如 ReadLock）
         * */
        static final AbstractQueuedSynchronizer.Node SHARED = new AbstractQueuedSynchronizer.Node();
        /**
         *  表示线程以互斥的模式等待锁（如 ReetrantLock），互斥就是一 把锁只能由一个线程持有，不能同时存在多个线程使用同一个锁
         *  */
        static final AbstractQueuedSynchronizer.Node EXCLUSIVE = null;
    
        /**
         *值为 1，表示线程的获锁请求已经“取消”
         * */
        static final int CANCELLED =  1;
        /**
         * 值为-1，表示该线程一切都准备好了,就等待锁空闲出来给我
         * */
        static final int SIGNAL    = -1;
        /**
         *值为-2，表示线程等待某一个条件（Condition）被满足
         *  */
        static final int CONDITION = -2;
        /**
         * 值为-3，当线程处在“SHARED”模式时，该字段才会被使用 上
         */
        static final int PROPAGATE = -3;
    
        /**
         *该 int 变量表示线程在队列中的状态，其值就是上述提到的 CANCELLED、SIGNAL、CONDITION、PROPAGATE
         */
        volatile int waitStatus;
    
        /**
         *该变量类型为 Node 对象，表示该节点的前一个 Node 节点（前驱）
         */
        volatile AbstractQueuedSynchronizer.Node prev;
    
        /**
         * 该变量类型为 Node 对象，表示该节点的后一个 Node 节点（后继）
         */
        volatile AbstractQueuedSynchronizer.Node next;
    
        /**
         * 该变量类型为 Thread 对象，表示该节点的代表的线程
         */
        volatile Thread thread;
    
        /**
         * 该变量类型为 Node 对象，表示等待 condition 条件的 Node 节 点
         */
        AbstractQueuedSynchronizer.Node nextWaiter;
    
        /**
         * 如果节点在共享模式中等待，则返回true.
         */
        final boolean isShared() {
            return nextWaiter == SHARED;
        }
    
        /**
         * 返回前一个节点，如果为空则抛出NullPointerException。
         * @return the predecessor of this node
         */
        final AbstractQueuedSynchronizer.Node predecessor() throws NullPointerException {
            AbstractQueuedSynchronizer.Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }
    
        Node() {    // Used to establish initial head or SHARED marker
        }
    
        Node(Thread thread, AbstractQueuedSynchronizer.Node mode) {     // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
        }
    
        Node(Thread thread, int waitStatus) { // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }

5、 即其中包括了：

**（1）线程的 2 种等待模式：**

SHARED ：表示线程以共享的模式等待锁（如  ReadLock ） EXCLUSIVE：表示线程以互斥的模式等待锁（如  ReetrantLock ），互斥就是一 把锁只能由一个线程持有，不能同时存在多个线程使用同一个锁

**（2）线程在队列中的状态枚举：**

CANCELLED ：值为  1 ，表示线程的获锁请求已经“取消”

SIGNAL ：值为 \-1 ，表示该线程一切都准备好了 , 就等待锁空闲出来给我

CONDITION ：值为 \-2 ，表示线程等待某一个条件（ Condition ）被满足

PROPAGATE ：值为 \-3 ，当线程处在“ SHARED ”模式时，该字段才会被使用 上

初始化  Node  对象时，默认为  0

**（3）成员变量：**

waitStatus ：该  int  变量表示线程在队列中的状态，其值就是上述提到的 CANCELLED、 SIGNAL 、 CONDITION 、 PROPAGATE

prev ：该变量类型为  Node  对象，表示该节点的前一个  Node  节点（前驱）

next ：该变量类型为  Node  对象，表示该节点的后一个  Node  节点（后继）

thread ：该变量类型为  Thread  对象，表示该节点的代表的线程

nextWaiter ：该变量类型为  Node  对象，表示等待  condition  条件的  Node  节 点当前线程获取同步状态失败时，同步器会将当前线程以及等待状态等信息构 造成为一个节点（Node ）并将其加入同步队列，同时会阻塞当前线程，当同步 状态释放时，会把首节点中的线程唤醒，使其再次尝试获取同步状态。同步队列 中的节点（Node ）用来保存获取同步状态失败的线程引用、等待状态以及前驱 和后继节点。

6、head  和  tail

AQS  还拥有 **首节点** （ head ）和 **尾节点** （ tail ）两个引用，一个指向 **队列头节**  **点** ，而另一个指向 **队列尾节点** 。

*注意：因为首节点*  *head*  *是不保存线程信息的节点，仅仅是因为数据结构设*  *计上的需要，在数据结构上，这种做法往往叫做“空头节点链表”。对应的就有*  *“非空头结点链表”*

**二、节点在同步队列中的增加和移出**

***1、节点加入到同步队列***

当一个线程成功地获取了同步状态（或者锁），其他线程将无法获取到同步 状态，也就是获取同步状态失败，AQS  会将这个线程以及等待状态等信息构造成 为一个节点（Node ）并将其加入同步队列的尾部。而这个加入队列的过程必须 要保证线程安全，因此同步器提供了一个基于  CAS  的设置尾节点的方法： compareAndSetTail(Node expect,Nodeupdate)，它需要传递当前线程“认为”的尾 节点和当前节点，只有设置成功后，当前节点才正式与之前的尾节点建立关联。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

***2、首节点的变化***

首节点是获取同步状态成功的节点，首节点的线程在释放同步状态时，将会 唤醒后继节点，而后继节点将会在获取同步状态成功时将自己设置为首节点。设 置首节点是通过获取同步状态成功的线程来完成的，由于只有一个线程能够成功 获取到同步状态，因此设置头节点的方法并不需要使用 CAS  来保证，它只需要将 首节点设置成为原首节点的后继节点并断开原首节点的 next  引用即可。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

***三、独占式同步状态获取与释放***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

1、获取锁源码

通过调用同步器的  acquire(int arg) 方法可以获取同步状态，主要完成了同步 状态获取、节点构造、加入同步队列以及在同步队列中自旋等待的相关工作，其 主要逻辑是：

public final void acquire(int arg) {
    /**
    * !tryAcquire(arg) 返回false,抢到锁,直接短路，不往下走；返回true未抢到锁，往下走。
    **/
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

首先调用自定义同步器实现的 tryAcquire(int arg)方法，该方法需要保证线程 安全的获取同步状态。 如果同步状态获取失败（tryAcquire 返回 false），则构造同步节点（独占式 Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过 addWaiter(Node node)方法将该节点加入到同步队列的尾部， 最后调用 acquireQueued(Node node,int arg)方法，使得该节点以“死循环” 的方式获取同步状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒 主要依靠前驱节点的出队或阻塞线程被中断来实现。 addWaiter(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) {
                node.prev = pred;//定义此节点的前面是尾节点，即此节点是尾节点后的节点，是新的尾节点
                if (compareAndSetTail(pred, node)) {//正式设置尾节点
                    pred.next = node;
                    return node;//设置成功后返回
                }
            }
            enq(node);//上面先快速尝试一次，如果设置失败走这里
            return node;
        }

将当前线程包装成 Node 后，队列不为空的情况下，先尝试把当前节点加入 队列并成为尾节点，如果不成功或者队列为空进入 enq(final Node node)方法。

private Node enq(final Node node) {
            for (;;) {//死循环，自旋
                Node t = tail;
                if (t == null) { //首次 创建新的节点队列
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {//设置尾节点
                        t.next = node;
                        return t;
                    }
                }
            }
        }

在 enq(final Node node)方法中，同步器通过“死循环”来保证节点的正确添 加，这个死循环中，做了两件事，第一件，如果队列为空，初始化队列，new 出 一个空节点，并让**首节点**（head）和**尾节点**（tail）两个引用都指向这个空节点； 第二件事，把当前节点加入队列。 在“死循环”中只有通过 CAS 将节点设置成为尾节点之后，当前线程才能从 该方法返回，否则，当前线程不断地尝试设置。 节点进入同步队列之后，观察 acquireQueued(Node node,int arg)方法

final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {//此处自旋不会太耗费CPU,因为首次失败后会进入阻塞状态
                    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)方法中，当前线程在“死循环”中 尝试获取同步状态，而只有前驱节点是头节点才能够尝试获取同步状态，这是为 什么？原因有两个。

第一，头节点是成功获取到同步状态的节点，而头节点的线程释放了同步状 态之后，将会唤醒其后继节点，后继节点的线程被唤醒后需要检查自己的前驱节 点是否是头节点。

第二，维护同步队列的  FIFO  原则。 当前线程获取到同步状态后，让**首节点** （ head ）这个引用指向自己所在节点。 当同步状态获取成功后，当前线程就从 acquire  方法返回了。如果同步器实现的 是锁，那就代表当前线程获得了锁。

2、释放锁源码：

当前线程获取同步状态并执行了相应逻辑之后，就需要释放同步状态，使得 后续节点能够继续获取同步状态。通过调用同步器的 release(int arg) 方法可以释 放同步状态，该方法在释放了同步状态之后，会唤醒其后继节点（进而使后继节 点重新尝试获取同步状态）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

public final boolean release(int arg) {
            if (tryRelease(arg)) {//释放锁，子类实现
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);//唤醒首节点（head）所指向节点的后继节点线程
                return true;
            }
            return false;
        }

该方法执行时，会唤醒**首节点**（head）所指向节点的后继节点线程， unparkSuccessor(Node node)方法使用 LockSupport 来唤醒处于等待状态的线程。 而在 unparkSuccessor 中，

private void unparkSuccessor(Node node) {
            
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);//设置状态
    
            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);//唤head 指向节点的后继节点线程
        }

这段代码的意思，一般情况下，被唤醒的是 head 指向节点的后继节点线程， 如果这个后继节点处于被 cancel 状态，（我推测开发者的思路这样的：后继节点 处于被 cancel 状态，意味着当锁竞争激烈时，队列的第一个节点等了很久（一直 被还未加入队列的节点抢走锁），包括后续的节点 cancel 的几率都比较大，所以） 先从尾开始遍历，找到最前面且没有被 cancel 的节点。

总结

在获取同步状态时，同步器维护一个同步队列，获取状态失败的线程都会被 加入到队列中并在队列中进行自旋；移出队列（或停止自旋）的条件是前驱节点 为头节点且成功获取了同步状态。在释放同步状态时，同步器调用 tryRelease(int arg)方法释放同步状态，然后唤醒  head  指向节点的后继节点。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/c7f04410716b49b1ad94e53323229445.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221120/522dba4511864f998dbc715ac75b0b9f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221120/ab5801696f87457d9176189737839f8b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221120/2c02dc28e652472ab273e3684fdd2529.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221120/c3add635b5534ad080e9758f9fbafe19.png