Java 并发：队列同步器是什么？

来自：泥瓦匠@bysocket.com

• 一、什么是 AQS 队列同步器
• 二、什么是 CLH 同步队列
• 三、小结

什么是 AQS ?

Java的内置锁一直都是备受争议的，在JDK 1.6之前，synchronized这个重量级锁其性能一直都是较为低下，虽然在1.6后，进行大量的锁优化策略,但是与Lock相比synchronized还是存在一些缺陷的：虽然synchronized提供了便捷性的隐式获取锁释放锁机制（基于JVM机制），但是它却缺少了获取锁与释放锁的可操作性，可中断、超时获取锁，且它为独占式在高并发场景下性能大打折扣。

在介绍Lock之前，我们需要先熟悉一个非常重要的组件，掌握了该组件JUC包下面很多问题都不在是问题了。该组件就是AQS。

AQS，AbstractQueuedSynchronizer，即队列同步器。它是构建锁或者其他同步组件的基础框架（如ReentrantLock、ReentrantReadWriteLock、Semaphore等），JUC并发包的作者（Doug Lea）期望它能够成为实现大部分同步需求的基础。它是JUC并发包中的核心基础组件。

AQS解决了子啊实现同步器时涉及当的大量细节问题，例如获取同步状态、FIFO同步队列。基于AQS来构建同步器可以带来很多好处。它不仅能够极大地减少实现工作，而且也不必处理在多个位置上发生的竞争问题。

在基于AQS构建的同步器中，只能在一个时刻发生阻塞，从而降低上下文切换的开销，提高了吞吐量。同时在设计AQS时充分考虑了可伸缩行，因此J.U.C中所有基于AQS构建的同步器均可以获得这个优势。

AQS的主要使用方式是继承，子类通过继承同步器并实现它的抽象方法来管理同步状态。

AQS使用一个int类型的成员变量state来表示同步状态，当state>0时表示已经获取了锁，当state = 0时表示释放了锁。它提供了三个方法（getState()、setState(int newState)、compareAndSetState(int expect,int update)）来对同步状态state进行操作，当然AQS可以确保对state的操作是安全的。

AQS通过内置的FIFO同步队列来完成资源获取线程的排队工作，如果当前线程获取同步状态失败（锁）时，AQS则会将当前线程以及等待状态等信息构造成一个节点（Node）并将其加入同步队列，同时会阻塞当前线程，当同步状态释放时，则会把节点中的线程唤醒，使其再次尝试获取同步状态。

AQS主要提供了如下一些方法：

• getState()：返回同步状态的当前值；
• setState(intnewState)：设置当前同步状态；
• compareAndSetState(intexpect,intupdate)：使用CAS设置当前状态，该方法能够保证状态设置的原子性；
• tryAcquire(intarg)：独占式获取同步状态，获取同步状态成功后，其他线程需要等待该线程释放同步状态才能获取同步状态；
• tryRelease(intarg)：独占式释放同步状态；
• tryAcquireShared(intarg)：共享式获取同步状态，返回值大于等于0则表示获取成功，否则获取失败；
• tryReleaseShared(intarg)：共享式释放同步状态；
• isHeldExclusively()：当前同步器是否在独占式模式下被线程占用，一般该方法表示是否被当前线程所独占；
• acquire(intarg)：独占式获取同步状态，如果当前线程获取同步状态成功，则由该方法返回，否则，将会进入同步队列等待，该方法将会调用可重写的tryAcquire(int arg)方法；
• acquireInterruptibly(intarg)：与acquire(int arg)相同，但是该方法响应中断，当前线程为获取到同步状态而进入到同步队列中，如果当前线程被中断，则该方法会抛出InterruptedException异常并返回；
• tryAcquireNanos(intarg,longnanos)：超时获取同步状态，如果当前线程在nanos时间内没有获取到同步状态，那么将会返回false，已经获取则返回true；
• acquireShared(intarg)：共享式获取同步状态，如果当前线程未获取到同步状态，将会进入同步队列等待，与独占式的主要区别是在同一时刻可以有多个线程获取到同步状态；
• acquireSharedInterruptibly(intarg)：共享式获取同步状态，响应中断；
• tryAcquireSharedNanos(intarg,longnanosTimeout)：共享式获取同步状态，增加超时限制；
• release(intarg)：独占式释放同步状态，该方法会在释放同步状态之后，将同步队列中第一个节点包含的线程唤醒；
• releaseShared(intarg)：共享式释放同步状态；

那什么是 CLH同步队列？

在上面提到了AQS内部维护着一个FIFO队列，该队列就是CLH同步队列。

CLH同步队列是一个FIFO双向队列，AQS依赖它来完成同步状态的管理，当前线程如果获取同步状态失败时，AQS则会将当前线程已经等待状态等信息构造成一个节点（Node）并将其加入到CLH同步队列，同时会阻塞当前线程，当同步状态释放时，会把首节点唤醒（公平锁），使其再次尝试获取同步状态。

在CLH同步队列中，一个节点表示一个线程，它保存着线程的引用（thread）、状态（waitStatus）、前驱节点（prev）、后继节点（next），其定义如下：

static final class Node {
 
 /** 共享 */
 
 static final Node SHARED = new Node();
 


 
 /** 独占 */
 
 static final Node EXCLUSIVE = null;
 


 
 /**
 
     * 因为超时或者中断，节点会被设置为取消状态，被取消的节点时不会参与到竞争中的，他会一直保持取消状态不会转变为其他状态；
 
     */
 
 static final int CANCELLED = 1;
 


 
 /**
 
     * 后继节点的线程处于等待状态，而当前节点的线程如果释放了同步状态或者被取消，将会通知后继节点，使后继节点的线程得以运行
 
     */
 
 static final int SIGNAL    = -1;
 


 
 /**
 
     * 节点在等待队列中，节点线程等待在Condition上，当其他线程对Condition调用了signal()后，改节点将会从等待队列中转移到同步队列中，加入到同步状态的获取中
 
     */
 
 static final int CONDITION = -2;
 


 
 /**
 
     * 表示下一次共享式同步状态获取将会无条件地传播下去
 
     */
 
 static final int PROPAGATE = -3;
 


 
 /** 等待状态 */
 
 volatile int waitStatus;
 


 
 /** 前驱节点 */
 
 volatile Node prev;
 


 
 /** 后继节点 */
 
 volatile Node next;
 


 
 /** 获取同步状态的线程 */
 
 volatile Thread thread;
 


 
 Node nextWaiter;
 


 
 final boolean isShared() {
 
 return nextWaiter == SHARED;
 
 }
 


 
 final Node predecessor() throws NullPointerException {
 
 Node p = prev;
 
 if (p == null)
 
 throw new NullPointerException();
 
 else
 
 return p;
 
 }
 


 
 Node() {
 
 }
 


 
 Node(Thread thread, Node mode) {
 
 this.nextWaiter = mode;
 
 this.thread = thread;
 
 }
 


 
 Node(Thread thread, int waitStatus) {
 
 this.waitStatus = waitStatus;
 
 this.thread = thread;
 
 }
 
}
 

CLH同步队列结构图如下：

入列

学了数据结构的我们，CLH队列入列是再简单不过了，无非就是tail指向新节点、新节点的prev指向当前最后的节点，当前最后一个节点的next指向当前节点。代码我们可以看看addWaiter(Node node)方法：

 private Node addWaiter(Node mode) {
 
 //新建Node
 
 Node node = new Node(Thread.currentThread(), mode);
 
 //快速尝试添加尾节点
 
 Node pred = tail;
 
 if (pred != null) {
 
            node.prev = pred;
 
 //CAS设置尾节点
 
 if (compareAndSetTail(pred, node)) {
 
                pred.next = node;
 
 return node;
 
 }
 
 }
 
 //多次尝试
 
        enq(node);
 
 return node;
 
 }
 

addWaiter(Node node)先通过快速尝试设置尾节点，如果失败，则调用enq(Node node)方法设置尾节点

 private Node enq(final Node node) {
 
 //多次尝试，直到成功为止
 
 for (;;) {
 
 Node t = tail;
 
 //tail不存在，设置为首节点
 
 if (t == null) {
 
 if (compareAndSetHead(new Node()))
 
                    tail = head;
 
 } else {
 
 //设置为尾节点
 
                node.prev = t;
 
 if (compareAndSetTail(t, node)) {
 
                    t.next = node;
 
 return t;
 
 }
 
 }
 
 }
 
 }
 

在上面代码中，两个方法都是通过一个CAS方法compareAndSetTail(Node expect, Node update)来设置尾节点，该方法可以确保节点是线程安全添加的。在enq(Node node)方法中，AQS通过“死循环”的方式来保证节点可以正确添加，只有成功添加后，当前线程才会从该方法返回，否则会一直执行下去。

过程图如下：

出列

CLH同步队列遵循FIFO，首节点的线程释放同步状态后，将会唤醒它的后继节点（next），而后继节点将会在获取同步状态成功时将自己设置为首节点，这个过程非常简单，head执行该节点并断开原首节点的next和当前节点的prev即可，注意在这个过程是不需要使用CAS来保证的，因为只有一个线程能够成功获取到同步状态。过程图如下：

代码示例

本文示例读者可以通过查看下面仓库的中的 alibaba/java/ParentClass.java ：

• Github：https://github.com/JeffLi1993/java-core-learning-example
• Gitee：https://gitee.com/jeff1993/java-core-learning-example

如果您对这些感兴趣，欢迎 star、follow、收藏、转发给予支持！

参考资料

1. Doug Lea：《Java并发编程实战》
2. 方腾飞：《Java并发编程的艺术》