AQS详解

科技2022-07-13 158

前言：之前AQS、ReentrantLock、CountLatchDown大概原理都看懂了，面试的时候一问能说个大概，感觉这样也就行了，跟背课文一样背源码也没什么意思，但是根据被拒的情况可能只说个大概不能满足面试官，今天结合源码好好看一遍，适合有一定基础的同学。

参考：

https://blog.csdn.net/u013728021/article/details/87358517

https://www.cnblogs.com/waterystone/p/4920797.html

第一章 AQS简介

1.1 简要介绍

1.1.1 state标志

1.1.2 队列

1.2 支持的模式

1.3 需要用户实现的部分

第二章 AQS独占锁执行流程

2.1 aquire

2.2 release

第三章 AQS非独占锁执行流程

3.1 acquireShared

3.2 releaseShared

第一章 AQS简介

1.1 简要介绍

AQS，全称是 AbstractQueuedSynchronizer，中文译为抽象队列式同步器。

AQS 中有两个重要的东西，一个以Node为节点实现的链表的队列，还有一个用volatile修饰的STATE标志，并且通过CAS来改变它的值。

1.1.1 state标志

State的话一般用来标记一些状态，如在ReetrantLock中State用来标记是否有线程占有锁与重入次数，在CountLatchDown中用来标记占有锁的线程个数，这完全取决于用户。

AQS提供了三个方法来操作state

getState()setState()compareAndSetState()

1.1.2 队列

队列的话有以下特点：

链表结构，在头尾结点中，需要特别指出的是头结点是一个空对象结点，无任何意义，即傀儡结点；

每一个Node结点都维护了一个指向前驱的指针和指向后驱的指针，结点与结点之间相互关联构成链表；

入队在尾，出队在头，出队后需要激活该出队结点的后继结点，若后继结点为空或后继结点waitStatus>0，则从队尾向前遍历取waitStatus<0的触发阻塞唤醒；

队列中节点状态值（waitStatus，只能为以下值）

//常量：表示节点的线程是已被取消的 static final int CANCELLED = 1; //常量：表示当前节点的后继节点的线程需要被唤醒 static final int SIGNAL = -1; //常量：表示线程正在等待某个条件 static final int CONDITION = -2; //常量：表示下一个共享模式的节点应该无条件的传播下去 static final int PROPAGATE = -3;

1.2 支持的模式

AQS支持线程抢占两种锁——独占锁和共享锁：

独占锁：同一个时刻只能被一个线程占有，如ReentrantLock，ReentrantWriteLock等，它又可分为：公平锁：按照线程在队列中的排队顺序，先到者先拿到锁非公平锁：当线程要获取锁时，无视队列顺序直接去抢锁，谁抢到就是谁的共享锁：同一时间点可以被多个线程同时占有，如ReentrantReadLock，Semaphore等

AQS的所有子类中，要么使用了它的独占锁，要么使用了它的共享锁，不会同时使用它的两个锁。

1.3 需要用户实现的部分

isHeldExclusively()：该线程是否正在独占资源。只有用到condition才需要去实现它。 tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。 tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。 tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。 tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。

第二章 AQS独占锁执行流程

2.1 aquire

acquire是用来获取资源的，当我们自己实现的tryAcquire中返回true，即获取资源成功后，便不会执行多余的操作，会执行直接跳出，当获取资源失败后，会调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)这个函数。

public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }

我们把acquireQueued(addWaiter(Node.EXCLUSIVE), arg)拆解来看，对于addWaiter(Node.EXCLUSIVE)来说，如果尾部节点不是空的，就把新的线程封装成节点加入到尾部节点中，如果尾部节点是空的，说明当前队列是空的，则调用enq方法

private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; }

在enq方法中先建立头节点，再采用CAS方法把当前节点添加到尾部节点。

private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }

再之后我们分析acquireQueued方法，先判断当前直接节点的前驱节点是不是头节点，如果是头节点，那么说明这个节点是队列中第一个线程，那就再尝试获取锁，如果能获取锁，那就直接把当前线程当作当前线程，如果不能获取线程，那就调用shouldParkAfterFailedAcquire判断下这个节点能否被正常唤醒，如果能被正常，那么调用parkAndCheckInterrupt阻塞当前线程，release的阻塞到这里就结束了。

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 GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }

最后我们看一下它是如何判断能不能被正常唤醒的，当前置节点的SIGNAL等-1，会直接返回true，如果大于0，会删掉一直向前走，删掉SIGNAL大于0的节点，如果是其它值（如默认的0），则会把前置节点SIGNAL置为-1，后面两种SIGNAL的情况还会继续再循环一遍。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) /* * This node has already set status asking a release * to signal it, so it can safely park. */ return true; if (ws > 0) { /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */ do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }

2.2 release

首先判断调用tryRelease判断能不能释放，如果能释放的话将头节点传入unparkSuccessor()中

public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }

如果头节点状态是-1，修改头节点状态为0，找到头节点的下一个节点，如果下一个是空或者状态大于0(因为多线程吗，可能在增加删除节点时暂时出现这种情况)，则从队尾开始找，找到第一个状态是-1的节点，然后唤醒。

private void unparkSuccessor(Node node) { /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); /* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ 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); }

这里是本文最重要的：

当线程被LockSupport.unpark(s.thread)唤醒后，不是说它就会从release的地方去执行了，而是会到我们阻塞线程的地方去执行，也就是下面的代码处，它会从parkAndCheckInterrupt()后面继续执行，重新进入for循环，再次判断能不能获取（值得注意的是上面我们说到它会从后向前遍历，而这里还是增加了前面得节点是不是头结点的判断，所以本人推测从前或者从后遍历其实遍历的都是头节点的下一个节点）

第三章 AQS非独占锁执行流程

分析完独占锁，再看非独占锁就简单多了，两者很多内容都是相似的，共享锁这里主要是增加了一个当释放锁时唤醒的线程不只是队列中的第一个线程，而是队列中所有可能获取到资源的线程，只要其前面得节点状态是-1，那么就会唤醒令其尝试获取资源。

3.1 acquireShared

尝试获取共享资源，如果用户自己实现的tryAcquireShared返回值小于0，则表明获取共享资源失败，调用doAcquireShared进行阻塞。

public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0) doAcquireShared(arg); }

doAcquireShared中的内容相信大家看着也比较熟悉了，先把节点加入队列，如果当前节点的前一个节点是头节点，则重新尝试获取资源，如果能获取成功，则调用setHeadAndPropagate将当前节点设置为头节点，然后继续去唤醒后面的节点，如果当前节点不是头节点或者重新尝试获取资源失败了，则找到一个能被唤醒的位置，然后阻塞自己。

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 GC if (interrupted) selfInterrupt(); failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }

可以看到在setHeadAndPropagate中，首先调用setHead将当前节点设置为头节点，如果下一个节点满足条件且是共享锁，则调用doReleaseShared()去唤醒下一个节点。

private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below 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(); } }

在doReleaseShared()，获取头节点，如果头节点状态是-1，则用CAS将状态置为0并且唤醒头节点的下一个节点，下一个节点继续重复上面的流程，即尝试获取资源，获取资源成功则将自己置为头节点，然后唤醒后继节点。

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 changed break; } }

3.2 releaseShared

判断能否释放资源，如果能的话则唤醒后继节点，如果不能则结束，这个看懂了前面的doReleaseShared()应该就很好懂了。

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

Processed: 0.018, SQL: 8