JUC AbstractQueuedSynchronizer原理解析

古城微笑少年丶 2022-08-07 10:45 183阅读 0赞

注：文本是由网上资料整理修改而成，详见参考资料。

本文内容是基于jdk1.7.0\_76的，不同jdk版本可能某些实现细节会有所修改。

# 摘要    #

在java.util.concurrent包（下称j.u.c包）中，大部分的同步器（例如锁，屏障等等）都是基于AbstractQueuedSynchronizer类（下称AQS类），这个简单的框架而构建的。这个框架为同步状态的原子性管理、线程的阻塞和解除阻塞以及先进先出 (FIFO) 等待队列提供了一种通用的机制。

# 1. AQS设计与实现 #

AQS同步器，基本思想很简单，对外部提供下面两个操作：

**acquire：**

while(synchronization state does not allow acquire){
        enqueue current threadifnot already queued;
        possibly block current thread;
    }
    dequeue current threadifit was queued;

**release：**

update synchronization state;
    if(state may permit a blocked thread to acquire)
        unlock one or more queued threads;

要支持这两个操作，需要实现的三个条件：

*  Atomically managing synchronization state（同步状态的原子性管理）
 *  Blocking and unblocking threads（阻塞和唤醒线程）
 *  Maintaining queues（队列的管理）

## 1.1 同步状态 ##

AbstractQueuedSynchronizer类中的3个属性：

/**
         * Head of the wait queue, lazily initialized.  Except for
         * initialization, it is modified only via method setHead.  Note:
         * If head exists, its waitStatus is guaranteed not to be
         * CANCELLED.
         */
        private transient volatile Node head;
    
        /**
         * Tail of the wait queue, lazily initialized.  Modified only via
         * method enq to add new wait node.
         */
        private transient volatile Node tail;
    
        /**
         * The synchronization state.
         */
        private volatile int state;

AQS类使用单个int（32位）来保存同步状态，并暴露出getState、setState以及compareAndSetState操作来读取和更新这个状态。  
  
将同步状态限制为一个32位的整形是出于实践上的考量。虽然JSR166也提供了64位long字段的原子性操作，但这些操作在很多平台上还是使用内部锁的方式来模拟实现的，这会使同步器的性能可能不会很理想。当然，将来可能会有一个类是专门使用64位的状态的。然而现在就引入这么一个类到这个包里并不是一个很好的决定（译者注：JDK1.6中已经包含java.util.concurrent.locks.AbstractQueuedLongSynchronizer类，即使用 long 形式维护同步状态的一个 AbstractQueuedSynchronizer 版本）。目前来说，32位的状态对大多数应用程序都是足够的。在j.u.c包中，只有一个同步器类可能需要多于32位来维持状态，那就是CyclicBarrier类，所以，它用了锁（该包中大多数更高层次的工具亦是如此）。  
  
基于AQS的具体实现类必须根据暴露出的状态相关的方法定义tryAcquire和tryRelease方法，以控制acquire和release操作。当同步状态满足时，tryAcquire方法必须返回true，而当新的同步状态允许后续acquire时，tryRelease方法也必须返回true。这些方法都接受一个int类型的参数用于传递想要的状态(AQS的继承类可以随意的定义该参数的含义)。例如：可重入锁中，当某个线程从条件等待中返回，然后重新获取锁时，为了重新建立循环计数的场景。很多同步器并不需要这样一个参数，因此忽略它即可。

## 1.2 阻塞/唤醒 ##

在JSR166之前，阻塞线程和解除线程阻塞都是基于Java内置管程，没有其它非基于Java内置管程的API可以用来创建同步器。唯一可以选择的是Thread.suspend和Thread.resume，但是它们都有无法解决的竞态问题，所以也没法用：当一个非阻塞的线程在一个正准备阻塞的线程调用suspend前调用了resume，这个resume操作将不会有什么效果。  
  
j.u.c包有一个LockSuport类，这个类中包含了解决这个问题的方法。方法LockSupport.park阻塞当前线程除非/直到有个LockSupport.unpark方法被调用（unpark方法被提前调用也是可以的）。unpark的调用是没有被计数的，因此在一个park调用前多次调用unpark方法只会解除一个park操作。另外，它们作用于每个线程而不是每个同步器。一个线程在一个新的同步器上调用park操作可能会立即返回，因为在此之前可能有“剩余的”unpark操作。但是，在缺少一个unpark操作时，下一次调用park就会阻塞。虽然可以显式地消除这个状态（译者注：就是多余的unpark调用），但并不值得这样做。在需要的时候多次调用park会更高效。  
  
park方法同样支持可选的相对或绝对的超时设置，以及与JVM的Thread.interrupt结合 —— 可通过中断来unpark一个线程。一个线程被唤醒，有3种可能的原因：

*  Some other thread invokes unpark with the current thread as the target; （其他线程调用unpark唤醒了该线程）
 *  Some other thread interrupts the current thread;（其它线程中断了该线程）
 *  The call spuriously (that is, for no reason) returns.（没有理由...可能是操作系统调度时唤醒的，好任性！）

## 1.3 队列    ##

整个框架的关键就是如何管理被阻塞的线程的队列，该队列是严格的FIFO队列，因此，框架不支持基于优先级的同步。

同步队列的最佳选择是自身没有使用底层锁来构造的非阻塞数据结构，目前，业界对此很少有争议。而其中主要有两个选择：一个是Mellor-Crummey和Scott锁（MCS锁）[\[9\]][9]的变体，另一个是Craig，Landin和Hagersten锁（CLH锁）[\[5\]][5][\[8\]][8][\[10\]][10]的变体。一直以来，CLH锁仅被用于自旋锁。但是，在这个框架中，CLH锁显然比MCS锁更合适。因为CLH锁可以更容易地去实现“取消（cancellation）”和“超时”功能，因此我们选择了CLH锁作为实现的基础。但是最终的设计已经与原来的CLH锁有较大的出入，因此下文将对此做出解释。

### Node的数据结构 ###

![20150403161016064][]

WaitStatus –>节点的等待状态，一个节点可能位于以下几种状态：

*  CANCELLED = 1： 节点操作因为超时或者对应的线程被interrupt。节点不应该不留在此状态，一旦达到此状态将从CHL队列中踢出。
 *  SIGNAL = -1： 节点的继任节点是（或者将要成为）BLOCKED状态（例如通过LockSupport.park()操作），因此一个节点一旦被释放（解锁）或者取消就需要唤醒（LockSupport.unpack()）它的继任节点。
 *  CONDITION = -2：表明节点对应的线程因为不满足一个条件（Condition）而被阻塞。
 *  0： 正常状态，新生的非CONDITION节点都是此状态。

非负值标识节点不需要被通知（唤醒）。

### Node结点构成的队列    ###

![20150403161434451][]

CLH队列实际上并不那么像队列，因为它的入队和出队操作都与它的用途（即用作锁）紧密相关。它是一个双向链表构成的队列，通过两个字段`head`和`tail`来存取，这两个字段是可原子更新的。队列操作时从tail入队，而从head.next出队。其中head指向的是一个虚拟头结点，head指向的Node并不在真正的CLH队列里。head和tail在初始化时都指向了一个空节点，当第一次竞争出现时，才延迟初始化--创建一个结点作为dummy header，并设置另`tail=head`。

一个新的节点，node，通过一个原子操作入队：

do {
    	    pred = tail;
    	} while(!tail.compareAndSet(pred, node));

每一个节点的“释放”状态都保存在其前驱节点中。自旋后的出队操作只需将head字段指向刚刚得到锁的节点。

CLH锁的优点在于其入队和出队操作是快速、无锁的，以及无障碍的（即使在竞争下，某个线程总会赢得一次插入机会而能继续执行）；且探测是否有线程正在等待也很快（只要测试一下head是否与tail相等）；同时，“释放”状态是分散的（*译者注：几乎每个节点都保存了这个状态，当前节点保存了其后驱节点的“释放”状态，因此它们是分散的，不是集中于一块的。*），避免了一些不必要的内存竞争。

在原始版本的CLH锁中，节点间甚至都没有互相链接。自旋锁中，`pred`变量可以是一个局部变量。然而，Scott和Scherer证明了通过在节点中显式地维护前驱节点，CLH锁就可以处理“超时”和各种形式的“取消”：如果一个节点的前驱节点取消了，这个节点就可以滑动去使用前面一个节点的状态字段。

为了将CLH队列用于阻塞式同步器，需要做些额外的修改以提供一种高效的方式定位某个节点的后继节点。在自旋锁中，一个节点只需要改变其状态，下一次自旋中其后继节点就能注意到这个改变，所以节点间的链接并不是必须的。但在阻塞式同步器中，一个节点需要显式地唤醒（`unpark`）其后继节点。

AQS队列的节点包含一个`next`链接到它的后继节点。但是，由于没有针对双向链表节点的类似`compareAndSet`的原子性无锁插入指令，因此这个`next`链接的设置并非作为原子性插入操作的一部分，而仅是在节点被插入后简单地赋值：

pred.next = node;

`next`链接仅是一种优化。如果通过某个节点的`next`字段发现其后继结点不存在（或看似被取消了），总是可以使用`pred`字段从尾部开始向前遍历来检查是否真的有后续节点。

第二个对CLH队列主要的修改是将每个节点都有的状态字段用于控制阻塞而非自旋。在同步器框架中，仅在线程调用具体子类中的`tryAcquire`方法返回`true`时，队列中的线程才能从`acquire`操作中返回；而单个“released”位是不够的。但仍然需要做些控制以确保当一个活动的线程位于队列头部时，仅允许其调用`tryAcquire`；这时的`acquire`可能会失败，然后（重新）阻塞。这种情况不需要读取状态标识，因为可以通过检查当前节点的前驱是否为`head`来确定权限。与自旋锁不同，读取`head`以保证复制时不会有太多的内存竞争（ there is not enough memory contention reading head to warrant replication.）。然而，“取消”状态必须存在于状态字段中。

队列节点的状态字段也用于避免没有必要的`park`和`unpark`调用。在调用`park`前，线程设置一个“唤醒（signal me）”位，然后再一次检查同步和节点状态。一个释放的线程会清空其自身状态。这样线程就不必频繁地尝试阻塞，特别是在锁相关的类中，这样会浪费时间等待下一个符合条件的线程去申请锁，从而加剧其它竞争的影响。除非后继节点设置了“唤醒”位，否则这也可避免正在release的线程去判断其后继节点。这反过来也消除了这些情形：除非“唤醒”与“取消”同时发生，否则必须遍历多个节点来处理一个似乎为null的`next`字段。

抛开一些细节，基本的`acquire`操作的最终实现的一般形式如下（互斥，非中断，无超时）：

if(!tryAcquire(arg)) {
        node = create and enqueue new node;
        pred = node's effective predecessor;
        while (pred is not head node || !tryAcquire(arg)) {
            if (pred's signal bit is set)
                pard()
            else
                compareAndSet pred's signal bit to true;
            pred = node's effective predecessor;
        }
        head = node;
    }

`release`操作：

if(tryRelease(arg) && head node's signal bit is set) {
        compareAndSet head's bit to false;
        unpark head's successor, if one exist
    }

`acquire`操作的主循环次数依赖于具体实现类中 `tryAcquire`的实现方式。另一方面，在没有“取消”操作的情况下，每一个组件的 `acquire`和 `release`都是一个O(1)的操作，忽略 `park`中发生的所有操作系统线程调度。

支持“取消”操作主要是要在`acquire`循环里的`park`返回时检查中断或超时。由超时或中断而被取消等待的线程会设置其节点状态，然后`unpark`其后继节点。在有“取消”的情况下，判断其前驱节点和后继节点以及重置状态可能需要O(n)的遍历（n是队列的长度）。由于“取消”操作，该线程再也不会被阻塞，节点的链接和状态字段可以被快速重建。

## 1.4 条件队列 ##

AQS框架提供了一个`ConditionObject`类，给维护独占同步的类以及实现`Lock`接口的类使用。一个锁对象可以关联任意数目的条件对象，可以提供典型的管程风格的`await`、`signal`和`signalAll`操作，包括带有超时的，以及一些检测、监控的方法。

通过修正一些设计决策，`ConditionObject`类有效地将条件（conditions）与其它同步操作结合到了一起。该类只支持Java风格的管程访问规则，这些规则中，仅当当前线程持有锁且要操作的条件（condition）属于该锁时，条件操作才是合法的。这样，一个`ConditionObject`关联到一个`ReentrantLock`上就表现的跟内置的管程（通过`Object.wait`等）一样了。两者的不同仅仅在于方法的名称、额外的功能以及用户可以为每个锁声明多个条件。

`ConditionObject使用了与同步器一样的内部队列节点，即Node类，但是，是在一个单独的条件队列中维护这些节点的（Condition Queue没有用到Node.prev和Node.next指针，而是用Node.nextWaiter指针构造了一个单链表）。`

`signal`操作是通过将节点从条件队列转移到锁队列中来实现的，而没有必要在需要唤醒的线程重新获取到锁之前将其唤醒。

基本的`await`操作如下：

create and add new node to conditon queue;
    release lock;
    block until node is on lock queue;
    re-acquire lock;

`signal`操作如下：

transfer the first node from condition queue to lock queue;

因为只有在持有锁的时候才能执行这些操作，因此他们可以使用顺序链表队列操作来维护条件队列（在节点中用一个`nextWaiter`字段）。转移操作仅仅把第一个节点从条件队列中的链接解除，然后通过CLH插入操作将其插入到锁队列上。

实现这些操作主要复杂在，因超时或`Thread.interrupt`导致取消了条件等待时，该如何处理。“取消”和“唤醒”几乎同时发生就会有竞态问题，最终的结果遵照内置管程相关的规范。JSR133修订以后，就要求如果中断发生在`signal`操作之前，await方法必须在重新获取到锁后，抛出`InterruptedException`。但是，如果中断发生在`signal`后，`await`必须返回且不抛异常，同时设置线程的中断状态。

为了维护适当的顺序，队列节点状态变量中的一个位记录了该节点是否已经（或正在）被转移。“唤醒”和“取消”相关的代码都会尝试用`compareAndSet`修改这个状态。如果某次`signal`操作修改失败，就会转移队列中的下一个节点（如果存在的话）。如果某次“取消”操作修改失败，就必须中止此次转移，然后等待重新获得锁。

以下是AQS队列和Condition队列的出入结点的示意图，可以通过这几张图看出线程结点在两个队列中的出入关系和条件。

（1）在队列头的线程结点1被唤醒后获得了锁，于是该结点被设置为head（即移除了队列，因为head是虚拟结点）。然后执行了condition.await()，await()方法内部先创建新结点加入到了Condition队列尾部，之后释放了锁。然后线程1被阻塞。

lock.lock();
    			try {
    				while (test something)
    					condition.await();
    				//......
    			} finally {
    				lock.unlock();
    			}

![20150403164533269][]

（2）此时线程2获得锁，结点2被移除AQS队列并被设置为head。然后调用到了condition.signal()，signal()方法内部把结点4从Condition队列中移除，并唤醒线程4。之后执行了lock.unlock()释放了锁。

lock.lock();
    			try {
    				//......modify something
    				condition.signal();
    			} finally {
    				lock.unlock();
    			}

![20150403164548791][]

（3）线程4被唤醒后，仍然在condition.await()方法体中（线程4调用情况同线程1），于是重新获取锁（获取不到就入AQS队列然后阻塞）。终于在后来的某一时间，线程4获取锁成功，于是await()方法执行完毕。于是再次进行while()语句进行判断，如果判断成功，则再次调用await()，或者结束while循环，代码段执行结束，释放锁。

注意：该例子中使用while()condition.await()而不是if()condition.await()，原因线程解除阻塞，不一定是被其它线程unpack()唤醒的（前面有提到，可能性有3种），所以需要再次判断条件是否满足。

![20150403164607714][]

# 参考资料 #

[The java.util.concurrent Synchronizer Framework 中文翻译版][The java.util.concurrent Synchronizer Framework] (译文原文是并发大师Doug Lea的论文，主要描述了AQS这个框架基本原理、设计、实现、用法以及性能)

[JUC (Java Util Concurrency) 基础内容概述][JUC _Java Util Concurrency_] (介绍了AQS实现原理)

# AQS**源码解析** #

另外，这里链接一些AQS**源码解析**的文章：

[深度解析Java8 – AbstractQueuedSynchronizer的实现分析（上）][Java8 _ AbstractQueuedSynchronizer]（从ReentrantLock出发，完整的分析了AQS独占功能的API及内部实现）

[深度解析Java8 – AbstractQueuedSynchronizer的实现分析（下）][Java8 _ AbstractQueuedSynchronizer 1]（从CountDownLatch入手，深入分析了AQS关于共享锁方面的实现方式）

[AbstractQueuedSynchronizer的介绍和原理分析][AbstractQueuedSynchronizer]（有较详细的图文解释，并且附有继承AQS实现简单独占锁和共享锁的例子）

[9]: http://ifeve.com/aqs-2/#r9
[5]: http://ifeve.com/aqs-2/#r5
[8]: http://ifeve.com/aqs-2/#r8
[10]: http://ifeve.com/aqs-2/#r10
[20150403161016064]: /images/20220805/866fb84300be4a9a871e3e28a53a04b4.png
[20150403161434451]: /images/20220805/61797d9009524877b4990760ce7f2043.png
[20150403164533269]: /images/20220805/0449cef02a704743aa47239b61349e9c.png
[20150403164548791]: /images/20220805/257d5e99de814afe80ba2f2fecfd79e4.png
[20150403164607714]: /images/20220805/02b3078a22104f7288df99dcd11b8e36.png
[The java.util.concurrent Synchronizer Framework]: http://ifeve.com/aqs/
[JUC _Java Util Concurrency_]: http://my.oschina.net/lifany/blog/146699
[Java8 _ AbstractQueuedSynchronizer]: http://ifeve.com/jdk1-8-abstractqueuedsynchronizer/
[Java8 _ AbstractQueuedSynchronizer 1]: http://ifeve.com/jdk1-8-abstractqueuedsynchronizer-part2/
[AbstractQueuedSynchronizer]: http://ifeve.com/introduce-abstractqueuedsynchronizer/