一文搞懂到底什么是 AQS

媒介

日常开辟中，我们经常利用锁大概其他同步器来控制并发，那么它们的基础框架是什么呢？如何实现的同步功能呢？本文将详细讲授构建锁和同步器的基础框架--AQS，并根据源码分析其原理。
一、什么是 AQS？

1. AQS 简介

AQS（Abstract Queued Synchronizer），抽象队列同步器，它是用来构建锁或其他同步器的基础框架。虽然大多数步伐员可能永远不会利用到它，但是知道 AQS 的原理有助于明白一些锁或同步器的是如何运行的。
那么有哪些同步器是基于 AQS 实现的呢？这里仅是简单介绍，详情后续会单独总结一篇文章。
同步器说明CountDownLatch递减的计数器，直至所有线程的使命都执行完毕，才继续执行后续使命。Semaphore信号量，控制同时访问某个资源的数目。CyclicBarrier递增的计数器，所有线程达到屏障时，才会继续执行后续使命。ReentrantLock防止多个线程同时访问共享资源，类似 synchronized 关键字。ReentrantReadWriteLock维护了读锁和写锁，读锁允许多线程访问，读锁阻塞所有线程。Condition提供类似 Object 监视器的方法，于 Lock 共同可以实现等候/关照模式。FutureTask当一个线程需要等候另一线程把某个使命执行完后才能继续执行，此时可以利用 FutureTask如果你明白了 AQS 的原理，也可以基于它去自定义一个同步组件，下文会介绍。
2. AQS 数据布局

AQS 焦点是通过对同步状态的管理，来完成线程同步，底层是依靠一个双端队列来完成同步状态的管理。

• 当火线程获取同步状态失败后，会构造成一个 Node 节点并参加队列末尾，同实阻塞线程。
  
• 当同步状态释放时，会把头节点中的线程唤醒，让其再次尝试获取同步状态
  

如下图，这里只是简单绘制，详细流程见下面原理分析：

这里的每个 Node 节点都存储着当火线程、等候信息等。
3. 资源共享模式

我们在获取共享资源时，有两种模式：
模式说明示例独占模式Exclusive，资源同一时刻只能被一个线程获取ReentrantLock共享模式Share，资源可同时被多个线程获取Semaphore、CountDownLatch二、AQS 原理分析

先简单说下原理分析的流程：

• 同步状态相关源码；
  
• 须重写的方法；
  
• Node 节点布局分析；
  
• 独占模式下的同步状态的获取与释放；
  
• 共享模式下的同步状态的获取与释放；
  

1. 同步状态相关

上面介绍到， AQS 焦点是通过对同步状态的管理，来完成线程同步，所以起首介绍管理同步状态的三个方法，在自定义同步组件时，需要通过它们获取和修改同步状态。

1. //保证可见性
2. private volatile int state
4. //获取当前同步状态。
5. protected final int getState() {
6.     return state;
7. }
9. //设置当前同步状态。
10. protected final void setState(int newState) {
11.     state = newState;
12. }
14. //使用 CAS 设置当前状态，保证原子性。
15. protected final boolean compareAndSetState(int expect, int update) {
16.     // See below for intrinsics setup to support this
17.     return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
18. }

复制代码

2. 须重写的方法

AQS 是基于模板方法模式的，通过第一个 abstract 也可知道，AQS 是个抽象类，利用者需要继承 AQS 并重写指定方法。
以下这些方式是没有详细实现的，需要在利用 AQS 时在子类中去实现详细方法，比及介绍一些同步组件时，会详细说明如何重写。

1. //独占式获取同步状态，实现该方法须查询并判断当前状态是否符合预期，然后再进行CAS设置状态。
2. protected boolean tryAcquire (int arg)
4. //独占式释放同步状态，等待获取同步状态的线程将有机会获取同步状态。
5. protected boolean tryRelease (int arg)
7. //共享式获取同步状态，返回大于0的值表示获取成功，反之获取失败。
8. protected int tryAcquireShared (int arg)
10. //共享式释放同步状态。
11. protected boolean tryReleaseShared (int arg)
13. //当前同步器是否再独占模式下被线程占用，一般用来表示是否被当前线程独占。
14. protected boolean isHeldExclusively ()

复制代码

3. Node 源码

Node 是双端队列中的节点，是数据布局的告急部门，线程相关的信息都存在每一个 Node 中。

3.1 Node 布局源码

源码如下：

1. static final class Node {
2.     //标记当前节点的线程在共享模式下等待。
3.     static final Node SHARED = new Node();
4.    
5.     //标记当前节点的线程在独占模式下等待。
6.     static final Node EXCLUSIVE = null;
7.    
8.     //waitStatus的值，表示当前节点的线程已取消（等待超时或被中断）
9.     static final int CANCELLED =  1;
10.    
11.     //waitStatus的值，表示后继节点的线程需要被唤醒
12.     static final int SIGNAL    = -1;
13.    
14.     //waitStatus的值，表示当前节点在等待某个条件，正处于condition等待队列中
15.     static final int CONDITION = -2;
16.    
17.     //waitStatus的值，表示在当前有资源可用，能够执行后续的acquireShared操作
18.     static final int PROPAGATE = -3;
20.     //等待状态，值如上，1、-1、-2、-3。
21.     volatile int waitStatus;
22.    
23.     //前趋节点
24.     volatile Node prev;
26.     //后继节点
27.     volatile Node next;
28.    
29.     //当前线程
30.     volatile Thread thread;
31.    
32.     //等待队列中的后继节点，共享模式下值为SHARED常量
33.     Node nextWaiter;
34.    
35.     //判断共享模式的方法
36.     final boolean isShared() {
37.         return nextWaiter == SHARED;
38.     }
39.    
40.     //返回前趋节点，没有报NPE
41.     final Node predecessor() throws NullPointerException {
42.         Node p = prev;
43.         if (p == null)
44.             throw new NullPointerException();
45.         else
46.             return p;
47.     }
49.     //下面是三个构造方法
50.     Node() {}    // Used to establish initial head or SHARED marke
51.    
52.     Node(Thread thread, Node mode) {     // Used by addWaiter
53.         this.nextWaiter = mode;
54.         this.thread = thread;
55.     }
56.     Node(Thread thread, int waitStatus) { // Used by Condition
57.         this.waitStatus = waitStatus;
58.         this.thread = thread;
59.     }
60. }

复制代码

3.2 设置头尾节点

Unsafe 类中，提供了一个基于 CAS 的设置头尾节点的方法，AQS 调用该方法进行设置头尾节点，包管并发编程中的线程安全。

1. //CAS自旋设置头节点
2. private final boolean compareAndSetHead(Node update) {
3.     return unsafe.compareAndSwapObject(this, headOffset, null, update);
4. }
7. //CAS自旋设置尾节点，expect为当前线程“认为”的尾节点，update为当前节点
8. private final boolean compareAndSetTail(Node expect, Node update) {
9.     return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
10. }
11.    

复制代码

4. 独占模式

资源同一时刻只能被一个线程获取，如 ReentrantLock。

4.1 获取同步状态

代码如下，调用 acquire 方法可以获取同步状态，底层就是调用须重写方法中的 tryAcquire。如果获取失败则进入同步队列中，即使后续对线程进行终端操作，线程也不会从同步队列中移除。

1. public final void acquire(int arg) {
2.     //调用须重写方法中的tryAcquire
3.     if (!tryAcquire(arg) &&
4.         //失败则进入同步队列中
5.         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
6.         selfInterrupt();
7. }

复制代码

获取失败会先调用 addWaiter 方法将当火线程封装成独占式模式的节点，添加到AQS的队列尾部，源码如下。

1. private Node addWaiter(Node mode) {
2.     //将当前线程封装成对应模式下的Node节点
3.     Node node = new Node(Thread.currentThread(), mode);
5.     Node pred = tail;//尾节点
6.     if (pred != null) {
7.         //双端队列需要两个指针指向
8.         node.prev = pred;
9.         //通过CAS方式
10.         if (compareAndSetTail(pred, node)) {
11.             //添加到队列尾部
12.             pred.next = node;
13.             return node;
14.         }
15.     }
16.     //等待队列中没有节点，或者添加队列尾部失败则调用end方法
17.     enq(node);
18.     return node;
19. }
21. //Node节点通过CAS自旋的方式被添加到队列尾部，直到添加成功为止。
22. private Node enq(final Node node) {
23.     //死循环，类似 while(1)
24.     for (;;) {
25.         Node t = tail;
26.         if (t == null) { // 须要初始化，代表队列的第一个元素
27.             if (compareAndSetHead(new Node()))
28.                 //头节点就是尾节点
29.                 tail = head;
30.         } else {
31.             //双端队列需要两个指针指向
32.             node.prev = t;
33.             //通过自旋放入队列尾部
34.             if (compareAndSetTail(t, node)) {
35.                 t.next = node;
36.                 return t;
37.             }
38.         }
39.     }
40. }

复制代码

此时，通过 addWaiter 已经将当火线程封装成独占模式的 Node 节点，并成功放入队列尾部。接下来会调用acquireQueued 方法在等候队列中排队。

1. final boolean acquireQueued(final Node node, int arg) {
2.     //获取资源失败标识
3.     boolean failed = true;
4.     try {
5.         //线程是否被中断标识
6.         boolean interrupted = false;
7.         //死循环，类似 while(1)
8.         for (;;) {
9.             //获取当前节点的前趋节点
10.             final Node p = node.predecessor();
12.             //前趋节点是head，即队列的第二个节点，可以尝试获取资源
13.             if (p == head && tryAcquire(arg)) {
14.                 //资源获取成功将当前节点设置为头节点
15.                 setHead(node);
16.                 p.next = null; // help GC，表示head节点出队列
17.                 failed = false;
18.                 return interrupted;
19.             }
20.             //判断当前线程是否可以进入waitting状态，详解见下方
21.             if (shouldParkAfterFailedAcquire(p, node) &&
22.                 parkAndCheckInterrupt())        //阻塞当前线程，详解见下方
23.                 interrupted = true;
24.         }
25.     } finally {
26.         if (failed)
27.             //取消获取同步状态，源码见下方的取消获取同步状态章节
28.             cancelAcquire(node);
29.     }
30. }
32. //将当前节点设置为头节点
33. private void setHead(Node node) {
34.     head = node;
35.     node.thread = null;
36.     node.prev = null;
37. }
39. //判断当前线程是否可以进入waitting状态
40. private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
41.     //获取前趋节点的等待状态，含义见上方Node结构源码
42.     int ws = pred.waitStatus;
43.     if (ws == Node.SIGNAL)        //表示当前节点的线程需要被唤醒
44.         return true;
45.     if (ws > 0) {        //表示当前节点的线程被取消
47.         //则当前节点一直向前移动，直到找到一个waitStatus状态小于或等于0的节点
48.         do {
49.             node.prev = pred = pred.prev;
50.         } while (pred.waitStatus > 0);
51.         //排在这个节点的后面
52.         pred.next = node;
53.     } else {
54.         //通过CAS设置等待状态
55.         compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
56.     }
57.     return false;
58. }
60. //阻塞当前线程
61. private final boolean parkAndCheckInterrupt() {
62.     //底层调用的UnSafe类的方法 park：阻塞当前线程, unpark：使给定的线程停止阻塞
63.     LockSupport.park(this);
64.     //中断线程
65.     return Thread.interrupted();
66. }

复制代码

acquireQueued 方法中，只有当前驱节点等于 head 节点时，才能够尝试获取同步状态，这时为什么呢？
因为 head 节点是占有资源的节点，它释放后才会唤醒它的后继节点，所以需要检测。还有一个缘故原由是因为如果遇到了非 head 节点的其他节点出队或因停止而从等候中唤醒，这时种情况则需要判定前趋节点是否为 head 节点，是才允许获取同步状态。
获取同步状态的整体流程图如下：

4.2 释放同步状态

调用须重写方法中的 tryAcquire 进行同步状态的释放，成功则唤醒队列中最前面的线程，详细如下。

1. public final boolean release(int arg) {
2.     //调用须重写方法中的tryRelease
3.     if (tryRelease(arg)) {
4.         Node h = head;
5.         if (h != null && h.waitStatus != 0)
6.             //唤醒后继节点的线程，详情见下方
7.             unparkSuccessor(h);
8.         return true;
9.     }
10.     return false;
11. }
13. //唤醒后继节点的线程
14. private void unparkSuccessor(Node node) {
15.     //获取当前节点的等待状态
16.     int ws = node.waitStatus;
17.     if (ws < 0)
18.         //小于0则，则尝试CAS设为0
19.         compareAndSetWaitStatus(node, ws, 0);
21.     //获取后继节点
22.     Node s = node.next;
24.     //后继节点为空或者等待状态大于0，代表被节点被取消
25.     if (s == null || s.waitStatus > 0) {
26.         s = null;
27.         //将队列中的所有节点都向前移动
28.         for (Node t = tail; t != null && t != node; t = t.prev)
29.             if (t.waitStatus <= 0)
30.                 s = t;
31.     }
32.     //不为空则进行唤醒操作
33.     if (s != null)
34.         //底层调用的UnSafe类的方法 park：阻塞当前线程, unpark：使给定的线程停止阻塞
35.         LockSupport.unpark(s.thread);
36. }

复制代码

5.2 释放同步状态

代码如下，调用 releaseShared 方法可以释放同步状态，底层就是先调用须重写方法中的 tryReleaseShared。

1. public final void acquireInterruptibly(int arg)
2.         throws InterruptedException {
3.     if (Thread.interrupted())//中断则抛出异常
4.         throw new InterruptedException();
5.     if (!tryAcquire(arg))
6.         doAcquireInterruptibly(arg);
7. }

复制代码

6. 取消获取同步状态

无论是独占模式照旧共享模式，所有的程获取同步状态的过程中，如果发生异常或是超时唤醒等，都需要将当前的节点出队，源码如下。
[code]//一般在获取同步状态方法的finally块中private void cancelAcquire(Node node) {    if (node == null)        return;    node.thread = null;                //当火线程节点设为null    Node pred = node.prev;                //获取前驱节点    //前趋节点为取消状态，向前遍历找到非取消状态的节点    while (pred.waitStatus > 0)        node.prev = pred = pred.prev;    Node predNext = pred.next;        //获取非取消节点的下一个节点    node.waitStatus = Node.CANCELLED;        //将当前节点的等候状态设为取消状态    //当前节点是尾节点，则自旋将尾节点设置为前一个非取消节点    if (node == tail && compareAndSetTail(node, pred)) {        //将尾节点设为前一个非取消的节点，并将其后继节点设为null，help GC        compareAndSetNext(pred, predNext, null);    } else {             int ws;//用于表示等候状态        //pred != head：前一个非取消的节点非头节点也非尾节点        //ws == Node.SIGNAL：当前等候状态为待唤醒        //若不是待唤醒则CAS设置为待唤醒状态        //前一个非取消的节点的线程不为null        if (pred != head &&            ((ws = pred.waitStatus) == Node.SIGNAL ||             (ws