导读：
我们日常开发中，经常会碰到并发的场景，在 Java 中语言体系里，我们会想到 ReentrantLock、CountDownLatch、Semaphore 等工具，但你是否清楚它们内部的实现原理？这些工具都很类似，底层都是基于 AbstractQueuedSynchronizer（AQS）来实现的。今天我们就来一起学习 AQS 内部原理。俗话说知己知彼百战百胜，如果我们了解了其中的原理，用该类工具开发就可以做到事半功倍。
文｜夏福利 网易智企资深开发工程师
一、AQS 执行框架
下图是 AQS 大体执行框架：
通过上面这张图能够了解到 AQS 大概的执行过程。ReentrantLock、CountDownLatch、Semaphore 都是在这个流程上封装的。
拿 ReentrantLock 来说，ReentrantLock 是独占锁， 可以是公平锁，也可以是非公平锁， 通过这张图可以很好理解了。
ReentrantLock 是独占锁体现在同一时刻只有一个线程能够尝试获取资源成功， 其他获取失败的都会加入阻塞队列的队尾进行排队。
公平锁就是线程严格按照阻塞队列的排列顺序获取资源，先到先得，不得插队。如下图所示：
而非公平锁就可能存在插队的可能。 例如，如果上面头节点被唤醒，正准备尝试获取资源，这时来了一个线程也尝试获取资源，有可能新来的线程获取资源成功，而头结点获取资源失败。这就是非公平锁。
在 ReentrantLock 源码中可以看到非公平锁会尝试获取资源时，不会考虑阻塞队列是否为空， 如果能够获取资源成功则直接占用了资源。获取失败才会加入阻塞队列。代码如下：
final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false;}
复制代码
而对于公平锁尝试获取资源时，会判断阻塞队列是否为空（和非公平锁关键差别所在），如下：
static final class FairSync extends Sync { private static final long serialVersionUID = -3000897897090466540L; final void lock() { acquire(1); } /** * Fair version of tryAcquire. Don't grant access unless * recursive call or no waiters or is first. */ @ReservedStackAccess protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }}
复制代码
二、AQS 实现原理详解
AQS 从名字上就知道是抽象类，通过模板方法定义了上面那张图的流程。对于“资源占用”和“资源释放”的定义，则是交给具体的子类去定义去实现。
对于共享锁，子类需要实现如下两个方法：
protected int tryAcquireShared(int arg)；protected boolean tryReleaseShared(int arg)；
复制代码
而对于独占锁，子类需要实现三个方法：
protected boolean tryAcquire(int arg)protected boolean tryRelease(int arg)protected boolean isHeldExclusively()
复制代码
isHeldExclusively： 该线程是否正在独占资源。只有用到 Condition 才需要去实现它；
tryAcquire： 独占方式。arg 为获取锁的次数，尝试获取资源，成功则返回 True，失败则返回 False；
tryRelease： 独占方式。arg 为释放锁的次数，尝试释放资源，成功则返回 True，失败则返回 False；
获取资源
首先，我们看一下独占锁获取资源的过程。
在 AQS 中，独占锁的获取资源的核心代码如下：
public final void acquire(int arg) { // 当 tryAcquire 返回 true 就说明获取到锁了，直接结束。 // 反之，返回 false 的话，就需要执行后面的方法。 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt();}
复制代码
如果子类的 tryAcquire 返回 true, 则表示获取锁成功，直接结束。
只要子类的 tryAcquire 方法返回 false，那么就说明获取锁失败，就需要将自己加入队列。
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; // 如果 tail 节点不是 null，就将新节点的 pred 节点设置为 tail 节点。 // 并且将新节点设置成 tail 节点。 if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } // 如果 tail 节点是 null，或者 CAS 设置 tail 失败。 // 在 enq 方法中处理 enq(node); return node;}
复制代码
如果 tail 节点为 null, 或者 CAS 设置 tail 失败，则通过自旋的方式加入尾结点。
private Node enq(final Node node) { for (;;) { Node t = tail; // 如果 tail 是 null，就创建一个虚拟节点，同时指向 head 和 tail，称为 初始化。 if (t == null) { // Must initialize if (compareAndSetHead(new Node())) tail = head; } else {// 如果不是 null // 和 上个方法逻辑一样，将新节点追加到 tail 节点后面，并更新队列的 tail 为新节点。 // 只不过这里是死循环的，失败了还可以再来 。 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } }}
复制代码
enq 方法的逻辑是什么呢？当 tail 是 null（没有初始化队列），就需要初始化队列了。CAS 设置 tail 失败，也会走这里，需要在 enq 方法中循环设置 tail。直到成功。
上面的过程用一张图表示如下：
将自己加入到阻塞队列后（注意 addWaiter 方法返回的是当前节点)，执行 acquireQueued() 方法，将当前节点对应的线程挂起，源码如下：
// 这里返回的节点是新创建的节点，arg 是请求的数量final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) { // 找上一个节点 final Node p = node.predecessor(); // 如果上一个节点是 head ，就尝试获取锁 // 如果 获取成功，就将当前节点设置为 head，注意 head 节点是永远不会唤醒的。 if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } // 在获取锁失败后，就需要阻塞了。 // shouldParkAfterFailedAcquire ---> 检查上一个节点的状态，如果是 SIGNAL 就阻塞，否则就改成 SIGNAL。 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); }}
复制代码
这个方法有两个逻辑：
先回答第二个问题：被唤醒之后做什么？
尝试拿锁，成功之后，将自己设置为 head，断开和 next 的连接。
再看第一个问题：如何将自己挂起？
具体逻辑在 shouldParkAfterFailedAcquire 方法中：
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; // 如果他的上一个节点的 ws 是 SIGNAL，他就需要阻塞。 if (ws == Node.SIGNAL) // 阻塞 return true; // 前任被取消。跳过前任并重试。 if (ws > 0) { do { // 将前任的前任 赋值给 当前的前任 node.prev = pred = pred.prev; } while (pred.waitStatus > 0); // 将前任的前任的 next 赋值为 当前节点 pred.next = node; } else { // 如果没有取消 || 0 || CONDITION || PROPAGATE，那么就将前任的 ws 设置成 SIGNAL. // 为什么必须是 SIGNAL 呢？ // 答：希望自己的上一个节点在释放锁的时候，通知自己（让自己获取锁） compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } // 重来 return false;}
复制代码
该方法的主要逻辑就是将前置节点的状态修改成 SIGNAL，告诉他：你释放锁的时候记得唤醒我。其中如果前置节点被取消了，就跳过他。那么在前置节点释放锁的时候，肯定会唤醒这个节点。
上面是独占锁获取资源过程，共享锁获取资源的过程类似，会有稍微的不同，核心代码如下:
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); } }
复制代码
这个方法同样是包含两个逻辑：
如何将自己挂起和独占锁没有区别。唤醒之后做什么是和独占锁区别的关键： 如果当前节点唤醒后获取到了锁后，会唤醒下一个节点。下一个节点唤醒后会继续唤醒下下一个节点，从而将所有等待共享锁的线程唤醒。核心代码如下：
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(); } }
复制代码
释放资源
上面讲了获取资源的逻辑，那如何释放资源呢？
同样，还是先看一下独占锁的释放逻辑：
public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; // 所有的节点在将自己挂起之前，都会将前置节点设置成 SIGNAL，希望前置节点释放的时候，唤醒自己。 // 如果前置节点是 0 ，说明前置节点已经释放过了。不能重复释放了，后面将会看到释放后会将 ws 修改成0. if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false;}
复制代码
从这个方法的判断就可以看出，head 必须不等于 0。为什么呢？上面获取资源过程中说到：当一个节点尝试挂起自己之前，都会将前置节点设置成 SIGNAL -1，就算是第一个加入队列的节点，在获取锁失败后，也会将初始化节点设置的 ws 设置成 SIGNAL。
而这个判断也是防止多线程重复释放，那么在释放锁之后，肯定会将 ws 状态设置成 0。防止重复操作。代码如下：
private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) // 将 head 节点的 ws 改成 0，清除信号。表示，他已经释放过了。不能重复释放。 compareAndSetWaitStatus(node, ws, 0); Node s = node.next; // 如果 next 是 null，或者 next 被取消了。就从 tail 开始向上找节点。 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);}
复制代码
唤醒之后的逻辑是什么样子的还记得吗？在上面讲解资源获取时有讲到：
线程唤醒后尝试拿锁，拿锁成功则设置自己为 head，断开前任 head 和自己的连接。
final boolean acquireQueued(final Node node, long arg) { boolean failed = true; try { boolean interrupted = false; //唤醒之后再次进行for循环，尝试获取锁，获取成功则将自己设置为头结点 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; } }
复制代码
再来看一下共享锁的释放逻辑，代码如下：
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }
复制代码
doReleaseShared() 代码如下：
private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { // 将 head 节点的 ws 改成 0，清除信号。表示，他已经释放过了。不能重复释放。 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; } }
复制代码
同样，唤醒后做什么呢？
线程唤醒后尝试拿锁，拿锁成功则设置自己为 head，断开前任 head 和自己的连接， 并唤醒下一个节点，代码如下：
private void doAcquireShared(long arg) { final Node node = addWaiter(Node.SHARED); boolean failed = true; try { boolean interrupted = false; //唤醒后再次for循环，尝试获取锁，获取锁成功，则设置自己为头结点， //并唤醒下一个结点，从而一直传播下去，将所有等待共享锁的线程唤醒 for (;;) { final Node p = node.predecessor(); if (p == head) { long 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; } }}
复制代码
为了证明共享锁唤醒时是一个接一个被唤醒，我们用一个 demo 来验证下，示例代码如下：
CountDownLatch countDownLatch = new CountDownLatch(1); Thread t1 = new Thread(() -> { try { TimeUnit.SECONDS.sleep(1); countDownLatch.await(); System.out.println("线程1被唤醒了"); } catch (InterruptedException e) { e.printStackTrace(); } }); Thread t2 = new Thread(() -> { try { TimeUnit.SECONDS.sleep(2); countDownLatch.await(); System.out.println("线程2被唤醒了"); } catch (InterruptedException e) { e.printStackTrace(); } }); Thread t3 = new Thread(() -> { try { TimeUnit.SECONDS.sleep(3); countDownLatch.await(); System.out.println("线程3被唤醒了"); } catch (InterruptedException e) { e.printStackTrace(); } }); t1.start(); t2.start(); t3.start(); TimeUnit.SECONDS.sleep(4); countDownLatch.countDown(); }
复制代码
上面示例代码中，线程 1、2、3 按顺序加入阻塞队列，当主线程调用 countDown() 时，此时会唤醒线程 1，线程 1 唤醒后会唤醒线程 2，线程 2 会唤醒线程 3。从运行结果可以看出确实如此：
从而证明了上面的推断。
三、总结
独占锁和共享锁在获取资源失败时，都会将自己加入阻塞队列的尾部，并将前一节点的 ws 设置为 SINGAL，告诉他：释放锁的时候记得唤醒我。
独占锁和共享锁的不同之处在于：节点被唤醒后，独占锁线程不会唤醒下一个节点（要唤醒必须主动释放锁，比如使用 ReentrantLock 最后要调用 release() 方法主动释放锁)。
而对于共享锁来说，只要一个节点被唤醒了，那就会继续唤醒下一个节点，下一个节点又会去唤醒下下一个节点，从而将所有等待共享锁的线程唤醒。
作者介绍
夏福利，网易智企资深开发工程师，主要负责网易七鱼在线智能客服的研发
评论