什么是 AQS ？

什么是 AQS ？

1.什么是AQS？

AQS是英文单词AbstractQueuedSynchronizer的缩写，翻译过来就是队列同步器。

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

2.AQS的实现方式？

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

3.AQS的原理

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

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

4.AQS的两种方式？

1）独占式，同一时刻仅有一个线程持有同步状态。

2）共享式与独占式的最主要区别在于同一时刻独占式只能有一个线程获取同步状态，而共享式在同一时刻可以有多个线程获取同步状态。例如读操作可以有多个线程同时进行，而写操作同一时刻只能有一个线程进行写操作，其他操作都会被阻塞。

 5.CLH同步队列

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

 //  CLH同步队列节点基本单元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();elsereturn p;}Node() {    // Used to establish initial head or SHARED marker
        }Node(Thread thread, Node mode) {     // Used by addWaiterthis.nextWaiter = mode;this.thread = thread;}Node(Thread thread, int waitStatus) { // Used by Conditionthis.waitStatus = waitStatus;this.thread = thread;}}

 CLH同步队列结构图:

 

入列（源码分析）：

/*** 先通过快速尝试设置尾节点，如果失败，则调用enq(Node node)方法设置尾节点* @param mode* @return*/private Node addWaiter(Node mode) {//新建NodeNode 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;}/*** AQS通过“死循环”的方式来保证节点可以正确添加，只有成功添加后，当前线程才会从该方法返回，否则会一直执行下去。* @param node* @return*/private Node enq(final Node node) {//自旋for (; ; ) {Node t = tail;//tail不存在，设置为首节点if (t == null) { //CAS设置头节点，原子性操作if (compareAndSetHead(new Node())) {tail = head;}} else {//设置为尾节点node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}}}

如图所示：

出列：

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

6.独占式同步状态的获取与释放

1）独占式获取同步状态（源码分析）：

/*** 模板方法，该方法为独占式获取同步状态，但是该方法对中断不敏感，也就是说由于线程获取同步状态失败加入到CLH同步队列中，* 后续对线程进行中断操作时，线程不会从同步队列中移除* @param arg*/public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {selfInterrupt();}}

 //去尝试获取锁，获取成功则设置锁状态并返回true，否则返回false。该方法自定义同步组件自己实现，该方法必须要保证线程安全的获取同步状态protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException();}

 //当前线程会根据公平性原则来进行阻塞等待（自旋）,直到获取锁为止；并且返回当前线程在等待过程中有没有中断过。final boolean acquireQueued(final Node node, int arg) {//是否获取同步状态成功boolean failed = true;try {//中断标志boolean interrupted = false;/** 自旋*/for (; ; ) {//当前线程的前驱节点final Node p = node.predecessor();//当前线程的前驱节点是头结点，且同步状态成功if (p == head && tryAcquire(arg)) {//当前线程为头节点
                    setHead(node);p.next = null; // help GCfailed = false;return interrupted;}//获取失败，线程等待if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) {interrupted = true;}}} finally {if (failed) {cancelAcquire(node);}}}

 /*** 设置当前线程中断*/static void selfInterrupt() {Thread.currentThread().interrupt();}

 

/*** 在自旋中，异常情况导致当前节点无法参与正常任务，则要进行取消，队列移除* 1。当前节点是尾部节点* 2。当前节点不是尾部节点* 取消正在进行尝试获取同步状态的节点*/private void cancelAcquire(Node node) {if (node == null) {return;}node.thread = null;//前驱节点Node pred = node.prev;//前驱节点状态 > 0 ，则为Cancelled,表明该节点已经超时或者被中断了，需要从同步队列中取消while (pred.waitStatus > 0) {node.prev = pred = pred.prev;}//前驱节点的后继节点Node predNext = pred.next;//设置当前节点的waitStatus为CANCELLEDnode.waitStatus = Node.CANCELLED;//如果当前节点是尾节点，并且设置前驱节点为尾节点成功if (node == tail && compareAndSetTail(node, pred)) {//原子操作compareAndSetNext(pred, predNext, null);} else {int ws;if (pred != head &&((ws = pred.waitStatus) == Node.SIGNAL ||(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&pred.thread != null) {Node next = node.next;if (next != null && next.waitStatus <= 0) {compareAndSetNext(pred, predNext, next);}} else {//唤醒后继节点
                unparkSuccessor(node);}node.next = node; // help GC
        }}/*** 唤醒后继节点*/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);}}

2）独占式获取同步状态响应中断（源码分析）：

  AQS提供了acquire(int arg)方法以供独占式获取同步状态，但是该方法对中断不响应，对线程进行中断操作后，该线程会依然位于CLH同步队列中等待着获取同步状态。为了响应中断，AQS提供了acquireInterruptibly(int arg)方法，该方法在等待获取同步状态时，如果当前线程被中断了，会立刻响应中断抛出异常InterruptedException。

/*** 独占式获取响应中断* @param arg* @throws InterruptedException*/public final void acquireInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted()) {throw new InterruptedException();}//获取同步状态if (!tryAcquire(arg)) {doAcquireInterruptibly(arg);}}/*** 执行独占式获取响应中断* @param arg* @throws InterruptedException*/private void doAcquireInterruptibly(int arg)throws InterruptedException {final Node node = addWaiter(Node.EXCLUSIVE);boolean failed = true;try {for (; ; ) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) {throw new InterruptedException();}}} finally {if (failed) {cancelAcquire(node);}}}

doAcquireInterruptibly(int arg)方法与acquire(int arg)方法仅有两个差别。

1.方法声明抛出InterruptedException异常。

2.在中断方法处不再是使用interrupted标志，而是直接抛出InterruptedException异常。

3）独占式超时获取同步状态（源码分析）：

/*** 独占式超时获取同步状态* @param arg* @param nanosTimeout* @return* @throws InterruptedException*/public final boolean tryAcquireNanos(int arg, long nanosTimeout)throws InterruptedException {if (Thread.interrupted()) {throw new InterruptedException();}return tryAcquire(arg) ||doAcquireNanos(arg, nanosTimeout);}private boolean doAcquireNanos(int arg, long nanosTimeout)throws InterruptedException {//nanosTimeout <= 0if (nanosTimeout <= 0L) {return false;}//超时时间final long deadline = System.nanoTime() + nanosTimeout;//新增Node节点final Node node = addWaiter(Node.EXCLUSIVE);boolean failed = true;try {//自旋for (;;) {final Node p = node.predecessor();//获取同步状态成功if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return true;}/** 获取失败，做超时、中断判断*///重新计算需要休眠的时间nanosTimeout = deadline - System.nanoTime();//已经超时，返回falseif (nanosTimeout <= 0L)return false;//如果没有超时，则等待nanosTimeout纳秒//注：该线程会直接从LockSupport.parkNanos中返回，//LockSupport为JUC提供的一个阻塞和唤醒的工具类，后面做详细介绍if (shouldParkAfterFailedAcquire(p, node) &&nanosTimeout > spinForTimeoutThreshold)LockSupport.parkNanos(this, nanosTimeout);//线程是否已经中断了if (Thread.interrupted())throw new InterruptedException();}} finally {if (failed) {cancelAcquire(node);}}}

 流程图如下：

 

4）独占式同步状态释放（源码分析）：

 /*** 独占式同步状态释放* @param arg* @return*/public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0) {//唤醒后继节点
                unparkSuccessor(h);}return true;}return false;}//模版方法protected boolean tryRelease(int arg) {throw new UnsupportedOperationException();}

 

5）独占式简单总结

在AQS中维护着一个FIFO的同步队列，当线程获取同步状态失败后，则会加入到这个CLH同步队列的对尾并一直保持着自旋。在CLH同步队列中的线程在自旋时会判断其前驱节点是否为首节点，

如果为首节点则不断尝试获取同步状态，获取成功则退出CLH同步队列。当线程执行完逻辑后，会释放同步状态，释放后会唤醒其后继节点。

 

7.共享式同步状态的获取与释放

1）共享式同步状态获取

/*** 共享式同步状态获取** @param arg*/public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0) {doAcquireShared(arg);}}//模版方法protected int tryAcquireShared(int arg) {throw new UnsupportedOperationException();}//自旋方式获取同步状态private void doAcquireShared(int arg) {//共享式节点final Node node = addWaiter(Node.SHARED);boolean failed = true;try {boolean interrupted = false;for (; ; ) {//前驱节点final Node p = node.predecessor();//如果其前驱节点，获取同步状态if (p == head) {//尝试获取同步int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCif (interrupted) {selfInterrupt();}failed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) {interrupted = true;}}} finally {if (failed) {cancelAcquire(node);}}}//设置头，并且如果是共享模式且propagate大于0，则唤醒后续节点。private void setHeadAndPropagate(Node node, int propagate) {Node h = head;setHead(node);if (propagate > 0 || h == null || h.waitStatus < 0 ||(h = head) == null || h.waitStatus < 0) {Node s = node.next;if (s == null || s.isShared()) {doReleaseShared();}}}//共享式释放同步状态private void doReleaseShared() {for (; ; ) {Node h = head;if (h != null && h != tail) {int ws = h.waitStatus;if (ws == Node.SIGNAL) {if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue;            // loop to recheck cases
                    unparkSuccessor(h);} else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue;                // loop on failed CAS
            }if (h == head)                   // loop if head changedbreak;}}

acquireShared(int arg)方法不响应中断，与独占式相似，AQS也提供了响应中断、超时的方法，分别是：acquireSharedInterruptibly(int arg)、tryAcquireSharedNanos(int arg,long nanos)。

2）共享式同步状态获取

public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;}

因为可能会存在多个线程同时进行释放同步状态资源，所以需要确保同步状态安全地成功释放，一般都是通过CAS和循环来完成的。

 

3）阻塞和唤醒线程

在线程获取同步状态时如果获取失败，则加入CLH同步队列，通过通过自旋的方式不断获取同步状态，但是在自旋的过程中则需要判断当前线程是否需要阻塞，其主要方法在acquireQueued()：

if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) {interrupted = true;}

 

/*** 根据前驱节点判断当前线程是否应该被阻塞** @param pred* @param node* @return*/private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {//前驱节点状态int ws = pred.waitStatus;//状态为signal，表示当前线程处于等待状态，直接返回trueif (ws == Node.SIGNAL) {return true;}//前驱节点状态 > 0 ，则为Cancelled,表明该节点已经超时或者被中断了，需要从同步队列中取消if (ws > 0) {do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {//前驱节点状态为Condition、propagate
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}

这段代码主要检查当前线程是否需要被阻塞，具体规则如下：

1. 如果当前线程的前驱节点状态为SINNAL，则表明当前线程需要被阻塞，调用unpark()方法唤醒，直接返回true，当前线程阻塞
2. 如果当前线程的前驱节点状态为CANCELLED（ws > 0），则表明该线程的前驱节点已经等待超时或者被中断了，则需要从CLH队列中将该前驱节点删除掉，直到回溯到前驱节点状态 <= 0 ，返回false
3. 如果前驱节点非SINNAL，非CANCELLED，则通过CAS的方式将其前驱节点设置为SINNAL，返回false

如果 shouldParkAfterFailedAcquire(Node pred, Node node) 方法返回true，则调用parkAndCheckInterrupt()方法阻塞当前线程：

//阻塞当前线程private final boolean parkAndCheckInterrupt() {LockSupport.park(this);//设置当前线程中断return Thread.interrupted();}

 

parkAndCheckInterrupt() 方法主要是把当前线程挂起，从而阻塞住线程的调用栈，同时返回当前线程的中断状态。其内部则是调用LockSupport工具类的park()方法来阻塞该方法。

 

当线程释放同步状态后，则需要唤醒该线程的后继节点：

 public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)//唤醒后继节点
                unparkSuccessor(h);return true;}return false;}

调用unparkSuccessor(Node node)唤醒后继节点：

 private void unparkSuccessor(Node node) {//当前节点状态int ws = node.waitStatus;//当前状态 < 0 则设置为 0if (ws < 0)compareAndSetWaitStatus(node, ws, 0);//当前节点的后继节点Node s = node.next;//后继节点为null或者其状态 > 0 (超时或者被中断了)if (s == null || s.waitStatus > 0) {s = null;//从tail节点来找可用节点for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t;}//唤醒后继节点if (s != null)LockSupport.unpark(s.thread);}

可能会存在当前线程的后继节点为null，超时、被中断的情况，如果遇到这种情况了，则需要跳过该节点，但是为何是从tail尾节点开始，而不是从node.next开始呢？原因在于node.next仍然可能会存在null或者取消了，所以采用tail回溯办法找第一个可用的线程。最后调用LockSupport的unpark(Thread thread)方法唤醒该线程。

 

8.LockSupport

作用：当需要阻塞或者唤醒一个线程的时候，AQS都是使用LockSupport这个工具类来完成的。LockSupport是用来创建锁和其他同步类的基本线程阻塞原语

 每个使用LockSupport的线程都会与一个许可关联，如果该许可可用，并且可在进程中使用，则调用park()将会立即返回，否则可能阻塞。如果许可尚不可用，则可以调用 unpark 使其可用。但是注意许可不可重入，也就是说只能调用一次park()方法，否则会一直阻塞。

LockSupport定义了一系列以park开头的方法来阻塞当前线程，unpark(Thread thread)方法来唤醒一个被阻塞的线程。如下：

 

park(Object blocker)方法的blocker参数，主要是用来标识当前线程在等待的对象，该对象主要用于问题排查和系统监控。

 

park方法和unpark(Thread thread)都是成对出现的，同时unpark必须要在park执行之后执行，当然并不是说没有不调用unpark线程就会一直阻塞，park有一个方法，它带了时间戳（parkNanos(long nanos)：为了线程调度禁用当前线程，最多等待指定的等待时间，除非许可可用）。

 

 

 

 

 

 

 

 

 

 

 

转载于:.html