JAVA AQS结构及其原理分析
引言
AQS，即AbstractQueuedSynchronizer, 队列同步器，它是Java并发用来构建锁和其他同步组件的基础框架。大多数开发者可能都不会直接使用AQS，标准同步器类的集合能够满足绝大多数情况的需求，但如果能了解标准同步器类的实现方式，那么对于理解它们的工作原理是非常有帮助的。
AQS是一个抽象类，一般是同步组件的静态内部类，通过组合的方式使用。AQS本身没有实现任何同步接口的，它仅仅只是定义了同步状态的获取和释放的方法来供自定义的同步组件的使用。
AQS原理简介
AQS维护一个共享资源state，通过内置的FIFO来完成获取资源线程的排队工作。（这个内置的同步队列称为"CLH"队列）。该队列由一个一个的Node结点组成，每个Node结点维护一个prev引用和next引用，分别指向自己的前驱和后继结点。AQS维护两个指针，分别指向队列头部head和尾部tail。
从图中可以看出它是一个双向链表。当线程获取资源失败（比如tryAcquire时试图设置state状态失败），会被构造成一个结点加入CLH队列中，同时当前线程会被阻塞在队列中（通过LockSupport.park实现，其实是等待态）。当持有同步状态的线程释放同步状态时，会唤醒后继结点，然后此结点线程继续加入到对同步状态的争夺中。
首先来看AQS最主要的三个成员变量：
1 private transient volatile Node head;
2
3 private transient volatile Node tail;
4
5 private volatile int state;
int型的变量state用来表示同步状态. head和tail分别是同步队列的头结点和尾结点。假设state=0表示同步状态可用（如果用于锁，则表示锁可用），state=1表示同步状态已被占用（锁被占用）,状态信息通过procted类型的getState，setState，compareAndSetState进行操作.
AQS支持两种同步方式：
独占式
共享式
这样方便使用者实现不同类型的同步组件，独占式如ReentrantLock，共享式如Semaphore，CountDownLatch，组合式的如ReentrantReadWriteLock。总之，AQS为使用提供了底层支撑，如何组装实现，使用者可以自由发挥。
同步器的设计是基于模板方法模式的，一般的使用方式是这样：
使用者继承AbstractQueuedSynchronizer并重写指定的方法。
将AQS组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。
AQS定义的以下可重写的方法：
protected boolean tryAcquire(int arg)
独占式获取同步状态，试着获取，成功返回true，反之为false
protected boolean tryRelease(int arg)
独占式释放同步状态，等待中的其他线程此时将有机会获取到同步状态；
protected int tryAcquireShared(int arg)
共享式获取同步状态，返回值大于等于0，代表获取成功；反之获取失败；
protected boolean tryReleaseShared(int arg)
共享式释放同步状态，成功为true，失败为false
protected boolean isHeldExclusively()
是否在独占模式下被线程占用。
自定义同步器
那么我们如何使用AQS呢，首先，我们需要去继承AbstractQueuedSynchronizer这个类，然后我们根据我们的需求去重写相应的方法，比如要实现一个独占锁，那就去重写tryAcquire，tryRelease方法，要实现共享锁，就去重写tryAcquireShared，tryReleaseShared；最后，在我们的组件中调用AQS中的模板方法就可以了，而这些模板方法是会调用到我们之前重写的那些方法的。也就是说，我们只需要很小的工作量就可以实现自己的同步组件，重写的那些方法，仅仅是一些简单的对于共享资源state的获取和释放操作，至于像是获取资源失败，线程需要阻塞之类的操作，自然是AQS帮我们完成了。
对于使用者来讲，我们无需关心获取资源失败，线程排队，线程阻塞/唤醒等一系列复杂的实现，这些都在AQS中为我们处理好了。我们只需要负责好自己的那个环节就好，也就是获取/释放共享资源state的姿势T_T。很经典的模板方法设计模式的应用，AQS为我们定义好顶级逻辑的骨架，并提取出公用的线程入队列/出队列，阻塞/唤醒等一系列复杂逻辑的实现，将部分简单的可由使用者决定的操作逻辑延迟到子类中去实现即可。
下面的例子是独占锁的实现方式，看代码：
1import java.util.concurrent.locks.AbstractQueuedSynchronizer;
2
3public class MyLock {
4
5
6 private final Sync sync = new Sync();
7
8 public void lock() {
9 sync.acquire(1);
10 }
11
12 public void unlock() {
13 sync.release(1);
14 }
15
16 public boolean isLocked() {
17 return sync.isHeldExclusively();
18 }
19
20
21 private static class Sync extends AbstractQueuedSynchronizer {
22 @Override
23 protected boolean tryAcquire(int arg) {
24
25 //首先判断状态是否等于=0,如果状态==0，就将status设置为1
26 if (compareAndSetState(0,1)) {
27 //将当前线程赋值给独占模式的onwer
28 setExclusiveOwnerThread(Thread.currentThread());
29 return true;
30 }
31
32 return false;
33
34 }
35
36 @Override
37 protected boolean tryRelease(int arg) {
38 if (getState() == 0) {
39 throw new IllegalMonitorStateException();
40 }
41 setExclusiveOwnerThread(null);
42 setState(0);
43 return true;
44 }
45
46 @Override
47 protected boolean isHeldExclusively() {
48 return getState() == 1;
49 }
50 }
51
52
53}
我们测试下我们写的自定义同步器，我们启用30个线程，每个线程对i自加10000次，同步正常的话，最终结果应为300000；测试代码如下：
1public class Demo {
2
3 private static CyclicBarrier barrier = new CyclicBarrier(31);
4
5 private static int count;
6
7 private static final MyLock myLock = new MyLock();
8
9
10 public static void main(String[] args) throws Exception{
11 //说明:我们启用30个线程，每个线程对i自加10000次，同步正常的话，最终结果应为300000；
12 //未加锁前
13 for(int i=0;i<30;i++){
14 Thread t = new Thread(new Runnable() {
15 @Override
16 public void run() {
17 for(int i=0;i<10000;i++){
18 increment1();//没有同步措施的a++；
19 }
20 try {
21 barrier.await();//等30个线程累加完毕
22 } catch (Exception e) {
23 e.printStackTrace();
24 }
25 }
26 });
27 t.start();
28 }
29 barrier.await();
30 System.out.println("没有加锁，count="+count);
31 //加锁后
32 barrier.reset();//重置CyclicBarrier
33 count=0;
34 for(int i=0;i<30;i++){
35 new Thread(new Runnable() {
36 @Override
37 public void run() {
38 for(int i=0;i<10000;i++){
39 increment2();//a++采用Mutex进行同步处理
40 }
41 try {
42 barrier.await();//等30个线程累加完毕
43 } catch (Exception e) {
44 e.printStackTrace();
45 }
46 }
47 }).start();
48 }
49 barrier.await();
50 System.out.println("加锁后，count="+count);
51 }
52
53
54
55 /**
56 * 没有同步措施的a++
57 * @return
58 */
59 public static void increment1(){
60 count++;
61 }
62 /**
63 * 使用自定义的Mutex进行同步处理的a++
64 */
65 public static void increment2(){
66 myLock.lock();
67 count++;
68 myLock.unlock();
69 }
70}
运行结果：
1没有加锁，count=260399
2加锁后，count=300000
从运行结果可以看出，我们写的自定义同步器的锁生效了。
AQS源码分析
AQS的实现依赖内部的同步队列（FIFO双向队列），如果当前线程获取同步状态失败，AQS会将该线程以及等待状态等信息构造成一个Node，将其加入同步队列的尾部，同时阻塞当前线程，当同步状态释放时，唤醒队列的头节点。
下面举例说下获取和释放同步状态的过程：
获取同步状态
假设线程A要获取同步状态（这里想象成锁，方便理解），初始状态下state=0,所以线程A可以顺利获取锁，A获取锁后将state置为1。在A没有释放锁期间，线程B也来获取锁，此时因为state=1，表示锁被占用，所以将B的线程信息和等待状态等信息构成出一个Node节点对象，放入同步队列，head和tail分别指向队列的头部和尾部（此时队列中有一个空的Node节点作为头点，head指向这个空节点，空Node的后继节点是B对应的Node节点，tail指向它），同时阻塞线程B(这里的阻塞使用的是LockSupport.park()方法)。后续如果再有线程要获取锁，都会加入队列尾部并阻塞。
释放同步状态
当线程A释放锁时，即将state置为0，此时A会唤醒头节点的后继节点（所谓唤醒，其实是调用LockSupport.unpark(B)方法），即B线程从LockSupport.park()方法返回，此时B发现state已经为0，所以B线程可以顺利获取锁，B获取锁后B的Node节点随之出队。
Node 节点
Node结点是AbstractQueuedSynchronizer中的一个静态内部类.
1static final class Node {
2 /** waitStatus值，表示线程已被取消（等待超时或者被中断）*/
3 static final int CANCELLED = 1;
4 /** waitStatus值，表示后继线程需要被唤醒（unpaking）*/
5 static final int SIGNAL = -1;
6 /**waitStatus值，表示结点线程等待在condition上，当被signal后，会从等待队列转移到同步到队列中 */
7 /** waitStatus value to indicate thread is waiting on condition */
8 static final int CONDITION = -2;
9 /** waitStatus值，表示下一次共享式同步状态会被无条件地传播下去
10 static final int PROPAGATE = -3;
11 /** 等待状态，初始为0 */
12 volatile int waitStatus;
13 /**当前结点的前驱结点 */
14 volatile Node prev;
15 /** 当前结点的后继结点 */
16 volatile Node next;
17 /** 与当前结点关联的排队中的线程 */
18 volatile Thread thread;
19 /** ...... */
20 }
独占式
我们看下acquire方法，lock方法一般会直接代理到acquire上
1 public final void acquire(int arg) {
2 if (!tryAcquire(arg) &&
3 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
4 selfInterrupt();
5 }
这段代码很短，看着不太容易理解，我们稍微做下转化：
1 public final void acquire(int arg) {
2 boolean hasAcquired = tryAcquire(arg);
3 if (! hasAcquired){
4 Node currentThreadNode=addWaiter(Node.EXCLUSIVE);
5 boolean interrupted = acquireQueued(currentThreadNode, arg);
6
7if (interrupted) {
8 selfInterrupt();
9 }
10
11 }
12 }
1.tryAcquire方法尝试获取锁，如果成功就返回，如果不成功，走到2.
2.把当前线程和等待状态信息构造成一个Node节点，并将结点放入同步队列的尾部
3.该Node结点在队列中尝试获取同步状态，若获取不到，则阻塞结点线程，直到被前驱结点唤醒或者被中断.
addWaiter
addWaiter方式主要是将同步状态失败的线程，构造成一个Node结点，添加到同步队列尾部
1 private Node addWaiter(Node mode) {
2 Node node = new Node(Thread.currentThread(), mode);
3 Node pred = tail;
4 if (pred != null) {
5 node.prev = pred;
6 if (compareAndSetTail(pred, node)) {
7 pred.next = node;
8 return node;
9 }
10 }
11 enq(node);
12 return node;
13 }
创建一个Node对象，Node中包含了当前线程和Node模式(这时是排他模式)。tail是AQS的中表示同步队列队尾的属性，刚开始为null，所以进行enq(node)方法，从字面可以看出这是一个入队操作，来看下具体入队细节：
1private Node enq(final Node node) {
2 for (;;) {
3 Node t = tail;
4 if (t == null) { // Must initialize
5 if (compareAndSetHead(new Node()))
6 tail = head;
7 } else {
8 node.prev = t;
9 if (compareAndSetTail(t, node)) {
10 t.next = node;
11 return t;
12 }
13 }
14 }
15 }
enq方法是一个死循环，本身没有锁，可以多个线程并发访问，假如某个线程进入方法，此时head, tail都为null, 进入if(t==null)区域，从方法名可以看出这里是用CAS的方式创建一个空的Node作为头结点，因为此时队列中只一个头结点，所以tail也指向它，第一次循环执行结束。注意这里使用CAS是防止多个线程并发执行到这儿时，只有一个线程能够执行成功，防止创建多个同步队列。
进行第二次循环时(或者是其他线程enq时)，tail不为null，进入else区域。将当前线程的Node结点(简称CNode)的prev指向tail，然后使用CAS将tail指向当前节点。
通过上面分析可知，AQS的写入是一种双向链表的插入操作，至此addWaiter分析完毕。
acquireQueued
看下acquireQueued方法:
1final boolean acquireQueued(final Node node, int arg) {
2 boolean failed = true;
3 try {
4 boolean interrupted = false;
5 for (;;) {//死循环
6 final Node p = node.predecessor();//找到当前结点的前驱结点
7 if (p == head && tryAcquire(arg)) {//如果前驱结点是头结点，才tryAcquire，其他结点是没有机会tryAcquire的。
8 setHead(node);//获取同步状态成功，将当前结点设置为头结点。
9 p.next = null; // 方便GC
10 failed = false;
11 return interrupted;
12 }
13 // 如果没有获取到同步状态，通过shouldParkAfterFailedAcquire判断是否应该阻塞，parkAndCheckInterrupt用来阻塞线程
14 if (shouldParkAfterFailedAcquire(p, node) &&
15 parkAndCheckInterrupt())
16 interrupted = true;
17 }
18 } finally {
19 if (failed)
20 cancelAcquire(node);
21 }
22 }
可以看到，acquireQueued方法也是一个死循环，直到进入 if (p == head && tryAcquire(arg))条件方法块。还是接着刚才的操作来分析。acquireQueued接收的参数是addWaiter方法的返回值。node.predecessor()返回当前节点的前置节点，在这里也就是head节点，所以p==head成立，进而进行tryAcquire操作，即争用锁, 如果获取成功，则进入if方法体，看下接下来的操作：
1) 将当前节点设置为头节点。
2) 将当前节点的前置节点设置的next设置为null。
上面操作即完成了FIFO的出队操作。
从上面的分析可以看出，只有队列的第二个节点可以有机会争用锁，如果成功获取锁，则此节点晋升为头节点。对于第三个及以后的节点，if (p == head)条件不成立，首先进行shouldParkAfterFailedAcquire(p, node)操作（争用锁失败的第二个节点也如此）.
shouldParkAfterFailedAcquire
shouldParkAfterFailedAcquire方法是判断一个争用锁的线程是否应该被阻塞
1private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
2 //获取前驱结点的wait值
3 int ws = pred.waitStatus;
4 if (ws == Node.SIGNAL)//若前驱结点的状态是SIGNAL，意味着当前结点可以被安全地park
5 return true;
6 if (ws > 0) {
7 // ws>0，只有CANCEL状态ws才大于0。若前驱结点处于CANCEL状态，也就是此结点线程已经无效，从后往前遍历，找到一个非CANCEL状态的结点，将自己设置为它的后继结点
8 do {
9 node.prev = pred = pred.prev;
10 } while (pred.waitStatus > 0);
11 pred.next = node;
12 } else {
13 // 若前驱结点为其他状态，将其设置为SIGNAL状态
14 compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
15 }
16 return false;
17 }
shouldParkAfterFailedAcquire方法是判断一个争用锁的线程是否应该被阻塞。它首先判断一个节点的前置节点的状态是否为Node.SIGNAL，如果是，是说明此节点已经将状态设置如果锁释放，则应当通知它，所以它可以安全的阻塞了，返回true。
如果前节点的状态大于0，即为CANCELLED状态时，则会从前节点开始逐步循环找到一个没有被“CANCELLED”节点设置为当前节点的前节点，返回false。在下次循环执行shouldParkAfterFailedAcquire时，返回true。这个操作实际是把队列中CANCELLED的节点剔除掉。
前节点状态小于0的情况是对应ReentrantLock的Condition条件等待的，这里不进行展开。
如果shouldParkAfterFailedAcquire返回了true，则会执行：“parkAndCheckInterrupt()”方法，它是通过LockSupport.park(this)将当前线程挂起到WATING状态，它需要等待一个中断、unpark方法来唤醒它，通过这样一种FIFO的机制的等待，来实现了Lock的操作。
release
当前线程执行完自己的逻辑之后，需要释放同步状态，来看看release方法的逻辑:
1public final boolean release(int arg) {
2 if (tryRelease(arg)) {//调用使用者重写的tryRelease方法，若成功，唤醒其后继结点，失败则返回false
3 Node h = head;
4 if (h != null && h.waitStatus != 0)
5 unparkSuccessor(h);//唤醒后继结点
6 return true;
7 }
8 return false;
9 }
在方法unparkSuccessor(Node)中，就意味着真正要释放锁了，它传入的是head节点（head节点是占用锁的节点），看下源码：
1private void unparkSuccessor(Node node) {
2 //获取wait状态
3 int ws = node.waitStatus;
4 if (ws < 0)
5 compareAndSetWaitStatus(node, ws, 0);// 将等待状态waitStatus设置为初始值0
6 Node s = node.next;//后继结点
7 if (s == null || s.waitStatus > 0) {//若后继结点为空，或状态为CANCEL（已失效），则从后尾部往前遍历找到一个处于正常阻塞状态的结点 进行唤醒
8 s = null;
9 for (Node t = tail; t != null && t != node; t = t.prev)
10 if (t.waitStatus <= 0)
11 s = t;
12 }
13 if (s != null)
14 LockSupport.unpark(s.thread);//使用LockSupprot唤醒结点对应的线程
15 }
内部首先会发生的动作是获取head节点的next节点，如果获取到的节点不为空，则直接通过：“LockSupport.unpark()”方法来释放对应的被挂起的线程，这样一来将会有一个节点唤醒后继续进入循环进一步尝试tryAcquire()方法来获取锁。
共享式源码简单分析
共享式：共享式地获取同步状态。对于独占式同步组件来讲，同一时刻只有一个线程能获取到同步状态，其他线程都得去排队等待，其待重写的尝试获取同步状态的方法tryAcquire返回值为boolean，这很容易理解；对于共享式同步组件来讲，同一时刻可以有多个线程同时获取到同步状态，这也是“共享”的意义所在。其待重写的尝试获取同步状态的方法tryAcquireShared返回值为int。
1 protected int tryAcquireShared(int arg) {
2 throw new UnsupportedOperationException();
3 }
1.当返回值大于0时，表示获取同步状态成功，同时还有剩余同步状态可供其他线程获取；
2.当返回值等于0时，表示获取同步状态成功，但没有可用同步状态了；
3.当返回值小于0时，表示获取同步状态失败。
acquireShared
1public final void acquireShared(int arg) {
2 if (tryAcquireShared(arg) < 0)//返回值小于0，获取同步状态失败，排队去；获取同步状态成功，直接返回去干自己的事儿。
3 doAcquireShared(arg);
4 }
doAcquireShared
1private void doAcquireShared(int arg) {
2 final Node node = addWaiter(Node.SHARED);//构造一个共享结点，添加到同步队列尾部。若队列初始为空，先添加一个无意义的傀儡结点，再将新节点添加到队列尾部。
3 boolean failed = true;//是否获取成功
4 try {
5 boolean interrupted = false;//线程parking过程中是否被中断过
6 for (;;) {//死循环
7 final Node p = node.predecessor();//找到前驱结点
8 if (p == head) {//头结点持有同步状态，只有前驱是头结点，才有机会尝试获取同步状态
9 int r = tryAcquireShared(arg);//尝试获取同步装填
10 if (r >= 0) {//r>=0,获取成功
11 setHeadAndPropagate(node, r);//获取成功就将当前结点设置为头结点，若还有可用资源，传播下去，也就是继续唤醒后继结点
12 p.next = null; // 方便GC
13 if (interrupted)
14 selfInterrupt();
15 failed = false;
16 return;
17 }
18 }
19 if (shouldParkAfterFailedAcquire(p, node) &&//是否能安心进入parking状态
20 parkAndCheckInterrupt())//阻塞线程
21 interrupted = true;
22 }
23 } finally {
24 if (failed)
25 cancelAcquire(node);
26 }
27 }
大体逻辑与独占式的acquireQueued差距不大，只不过由于是共享式，会有多个线程同时获取到线程，也可能同时释放线程，空出很多同步状态，所以当排队中的老二获取到同步状态，如果还有可用资源，会继续传播下去。
1private void setHeadAndPropagate(Node node, int propagate) {
2 Node h = head; // Record old head for check below
3 setHead(node);
4 if (propagate > 0 || h == null || h.waitStatus < 0) {
5 Node s = node.next;
6 if (s == null || s.isShared())
7 doReleaseShared();
8 }
9 }
releaseShared
releaseShared为释放同步状态
1public final boolean releaseShared(int arg) {
2 if (tryReleaseShared(arg)) {
3 doReleaseShared();//释放同步状态
4 return true;
5 }
6 return false;
7 }
doReleaseShared
1private void doReleaseShared() {
2 for (;;) {//死循环，共享模式，持有同步状态的线程可能有多个，采用循环CAS保证线程安全
3 Node h = head;
4 if (h != null && h != tail) {
5 int ws = h.waitStatus;
6 if (ws == Node.SIGNAL) {
7 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
8 continue;
9 unparkSuccessor(h);//唤醒后继结点
10 }
11 else if (ws == 0 &&
12 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
13 continue;
14 }
15 if (h == head)
16 break;
17 }
18 }
代码逻辑比较容易理解，需要注意的是，共享模式，释放同步状态也是多线程的，此处采用了CAS自旋来保证。
总结
AQS是JUC中很多同步组件的构建基础，简单来讲，它内部实现主要是状态变量state和一个FIFO队列来完成，同步队列的头结点是当前获取到同步状态的结点，获取同步状态state失败的线程，会被构造成一个结点（或共享式或独占式）加入到同步队列尾部（采用自旋CAS来保证此操作的线程安全），随后线程会阻塞；释放时唤醒头结点的后继结点，使其加入对同步状态的争夺中。
如果想学习Java工程化、高性能及分布式、深入浅出。性能调优、Spring，MyBatis，Netty源码分析的朋友可以加我的Java高级架构进阶群：180705916，群里有阿里大牛直播讲解技术，以及Java大型互联网技术的视频免费分享给大家
