|
|
前言 这篇文章写完放着也蛮久的了，今天终于发布了，对于拖延症患者来说也真是不容易～哈哈哈。
言归正传，其实吧。。我觉得对于大部分想了解 AQS 的朋友来说，明白 AQS 是个啥玩意儿以及为啥需要 AQS，其实是最重要的。就像我一开始去看 AQS 的时候，抱着代码就啃，看不懂就去网上搜。。但是网上文章千篇一律。。大部分都是给你逐行分析下代码然后就没了。。。
| 但其实对我们来说我知道为啥要这么干其实也相当重要。。 |
嗯。。所以就有了这篇文章。。笔者会先给你介绍下 AQS 的作者为啥要整这个东西。。然后笔者再结合自身感悟给你划了划重点。。如果你认真读了。。肯定会有所收获的哦
一、AQS 是什么？为什么需要 AQS ？ 试想有这么一种场景：有四个线程由于业务需求需要同时占用某资源，但该资源在同一个时刻只能被其中唯一线程所独占。那么此时应该如何标识该资源已经被独占，同时剩余无法获取该资源的线程又该何去何从呢？
 这里就涉及到了关于共享资源的竞争与同步关系。对于不同的开发者来说，实现的思路可能会有不同。这时如果能够有一个较为通用的且性能较优同步框架，那么可以在一定程度上帮助开发人员快速有效的完成多线程资源同步竞争方面的编码。
AQS 正是为了解决这个问题而被设计出来的。AQS 是一个集同步状态管理、线程阻塞、线程释放及队列管理功能与一身的同步框架。其核心思想是当多个线程竞争资源时会将未成功竞争到资源的线程构造为 Node 节点放置到一个双向 FIFO 队列中。被放入到该队列中的线程会保持阻塞直至被前驱节点唤醒。值得注意的是该队列中只有队首节点有资格被唤醒竞争锁。
从以下几个点切入翻阅 AQS 源码，那就相当如鱼得水了：
- 同步状态的处理
- FIFO 队列的设计，如何处理未竞争到资源的线程
- 竞争失败时线程如何处理
- 共享资源的释放
后面的章节主要会结合 AQS 源码，介绍下独占模式下锁竞争及释放相关内容。
二、同步状态的处理| private volatile int state; |
翻阅下 AQS 源码，不难发现有这么一个 volatile 类型的 state 变量。通俗的说这个 state 变量可以用于标识当前锁的占用情况。打个比方：当 state 值为 1 的时候表示当前锁已经被某线程占用，除非等占用的锁的线程释放锁后将 state 置为 0，否则其它线程无法获取该锁。这里的 state 变量用 volatile 关键字保证其在多线程之间的可见性。
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
} |
同时，我们发现 AQS 预留了个口子可以供开发人员按照自身需求进行二次重构。因此也就出现了类似与 ReentrantLock 可重入锁、 CountDownLatch 等实现。
三、AQS 灵魂队列的设计 对于整个 AQS 框架来说，队列的设计可以说重中之重。那么为什么 AQS 需要一个队列呢？
对于一个资源同步竞争框架来说，如何处理没有获取到锁的线程是非常重要的，比方说现在有 ABCD 四个线程同时竞争锁，其中线程 A 竞争成功了。那么剩下的线程 BCD 该咋办呢？
我们可以尝试试想下自己会如何解决：
- 线程自旋等待，不断重新尝试获取锁。这样虽然可以满足需求，但是众多线程同时自旋等待实际上是对 CPU 资源的一种浪费，这么做不太合适。
- 将线程挂起，等待锁释放时唤醒，再竞争获取。如果等待的线程比较多，同时被唤醒可能会发生“惊群”问题。
上面两种方法的可行性其实都不太高，对于一个同步框架来说，当有多个线程尝试竞争资源时，我们并不希望所有的线程同时来竞争锁。而且更重要的是，
。那么这个时候借助 FIFO 队列管理线程，既可以有效的帮助开发者监控线程，同时也可以在一定程度上减少饥饿问题出现的概率( 线程先入先出)。
除此之外 AQS 中用于存放线程的队列还有以下几点考量：
- Node 节点的设计
- 前驱、后继节点，分别保存当前节点在队列中的前驱节点和后继节点
- 节点状态：节点拥有不同的状态可以帮助我们更好的管理队列中的线程。在本文中我们只讨论 SIGNAL 和 CANCEL 状态。当前驱节点的状态为 SIGNAL 时，表示当前节点可以被安全挂起，锁释放时当前线程会被唤醒去尝试重新获取锁；CANCEL 状态表示当前线程被取消，无需再尝试获取锁，可以被移除队列
// 线程被取消
static final int CANCELLED = 1;
// 后续线程在锁释放后可以被唤醒
static final int SIGNAL = -1;
// 当前线程在 condition 队列中
static final int CONDITION = -2;
// 没有深入体会，表示下一次共享式同步状态获取将会无条件被传播下去
static final int PROPAGATE = -3; |
- AQS 中的双向线程队列 由于 Node 前驱和后继节点的存在。这里保存 Node 的队列实际上是一个双向队列。在这个队列里前驱节点的存在会更重要些：
| 当前新节点被插入到队列中时，如果前驱节点状态为取消状态。我们可以通过前驱节点不断往前回溯，完成一个类似滑动窗口的功能，跳过无效线程 |
，从而帮助我们更有效的管理等待队列中线程。而且上面也提过了，等待线程都放在队列中，一方面可以管控等待线程，另一方面也可以减少饥饿现象发生的概率。 - HEAD 和 TAIL HEAD 和 TAIL 节点分别指向队列的首尾节点。当第一次往队列中塞入一个新的节点时会构造一个虚拟节点作为 HEAD 头节点。为什么需要虚拟的 HEAD 头节点呢？因为在 AQS 的设计理念中，当前节点能够安心自我阻塞的前提条件是前驱节点在释放锁资源时，能够唤醒后继节点。
| 而插入到第一个队列中的节点，没有前驱节点怎么办，我们就构造一个虚拟节点来满足需求 |
。
同时 HEAD 和 TAIL 节点的存在加上双向队列的设计，整体的队列就显的非常灵活。
四、资源竞争(获取锁) 这一章节开始我们将结合源码对 AQS 获取锁的流程进行讨论。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
} |
acquire 方法用于获取锁，这里可以拆解为三步：
- tryAcquired： 看名字就知道用于尝试获取锁，并不保证一定可以获取锁，具体逻辑由子类实现。如果在这一步成功获取到了锁，后面的逻辑也就没有必要继续执行了。
- addWaiter：
| 尝试竞争锁资源失败后，我们就要考虑将这个线程构造成一个节点插入到队列中了 |
。这里的 addWaiter() 方法会将当前线程包装成一个 Node 节点后，维护到 FIFO 双向队列中。
private Node addWaiter(Node mode) {
// 将当前线程包装成一个 Node 节点
Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
// 如果 tail 节点不为空：新节点的前驱指向 tail，原尾节点的后继指向当前节点，当前节点成为新的尾节点
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
// 第一次往队列中新增节点时，会执行 enq 方法
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
// head 和 tail 在初始情况下都为 null
Node t = tail;
if (t == null) { // 初始化一个空节点用于帮助唤醒队列中的第一个有效线程
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 这段逻辑用于考虑多线程并发的场景，如果此时队列中已经有了节点
// 再次尝试将当前节点插至队尾
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
} |
这段逻辑不复杂：
- 当我们处理第一个节点时，此时 tail 节点为 null，因此会执行 enq() 方法。可以看到 enq 方法实际是一个死循环，只有当节点成功被插入到队列后，才能跳出去循环。那这么做的目的是什么呢？
| 其实不难看出，这里是为了应对多线程竞争而采取的妥协之策 |
。多个线程同时执行这段逻辑时，只有一个线程可以成功调用 compareAndSetHead() 并将 head 头指向一个新的节点，此时的 head 和 tail 都指向一个空节点。这个空节点的作用前面已经提过了，用于帮助后继节点可以在合适的场景下自我阻塞等待被唤醒。其它并发执行的线程执行 compareAndSetHead() 方法失败后，发现 tail 已经不为 null 了，依次将自己插入到 tail 节点后。 - 当 tail 节点不为空时，表示此时队列中有数据。因此我们借助 CAS 将新节点插入到尾节点之后，同时将 tail 指向新节点
 final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
// 前驱节点为 head 时，尝试获取锁
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);
}
} |
这里又是一个死循环
-
| 这里需要注意的是只有前驱节点为 head 时，我们才会再次尝试获取锁 |
。也就是在当前队列中，只有队首节点才会尝试获取锁。
。如果成功获取到了锁：将 node 节点设置为头节点，同时将前驱节点的 next 设置为 null 帮助 gc。 - 如果 node 节点前驱节点不为 head 或者获取锁失败，执行 shouldParkAfterFailedAcquire() 方法判断当前线程是否需要阻塞，如果需要阻塞则会调用 parkAndCheckInterrupt() 方法挂起当前线程
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* 当前驱节点状态为 SIGNAL 时，表示调用 release 释放前驱节点占用的锁时，
* 前驱会唤醒当前节点，可安全挂起当前线程等待被唤醒
*/
return true;
if (ws > 0) {
/*
* 前驱节点处于取消状态，我们需要跳过这个节点，并且重试
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// waitStatus 为 0 或 PROPAGATE 走的这里。后文会分析下什么时候 waitStatus 可能为 0
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
} |
| 当节点状态为 SIGNAL 时，表示当前线程可以被安全挂起。waitStats 大于0表示当前线程已经被取消，我们需要往前回溯找到有效节点。 |
在开始阅读这段代码时，一直想不通在哪些场景下 waitStatus 的状态可能为 0，在参阅了其它笔者分析的文章再加上自己的理解后，总结出以下两种场景：
-
| 当我们往队列中新插入一个节点时。队尾节点的 waitStatus 值应为初始状态 0 |
。此时执行 shouldParkAfterFailedAcquire() 方法会执行最后一个判断条件将前驱 waitStatus 状态更新为 SIGNAL,同时方法返回 false 。然后会继续执行一次 acquireQueued() 中的死循环，此时前驱节点的状态已经被更新为 SIGNAL，再次执行 shouldParkAfterFailedAcquire() 方法会返回 true，当前线程即可放心的将自己挂起，等待被线程唤醒。 -
| 当调用 release() 方法释放锁时，会将占用锁的节点的 waitStatus 状态更新为 0 |
，同时会调用 LockSupport.unpark() 方法唤醒后继节点。当后继节点被唤醒之后，会继续执行被挂起之前执行的 acquireQueued() 方法中的 for 循环再次尝试获取锁。
，如果获取不到锁则会再次执行 shouldParkAfterFailedAcquire() 方法。
为什么说被唤醒的线程不一定可以获取到锁呢？
对于基础的 acquire 方法来说，没有任何规则规定队首节点一定可以获取到锁。当我们在唤醒队列中的第一个有效线程时，此时如果出现了一个线程 A 尝试获取锁，那么该线程会调用 acquire() 方法尝试获取锁，如果运气不错，线程 A 完全有可能会窃取当前处于队列头中的线程获取锁的机会。
。那么为什么这么做？
如果队列头处于解除阻塞过程中，这一段时间实际上没有线程可以获取资源，属于一种资源浪费。所以这里只能认为是有一定概率的公平。
五、资源释放(释放锁) public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
private void unparkSuccessor(Node node) {
// 当状态小于 0 时，更新 waitStatus 值为 0
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
// 如果后继节点为 null 或者状态为取消，从尾结点向前查找状态不为取消的可用节点
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);
} |
release 整体流程比较简单。需要我们注意的就是为什么此时需要把 head 节点的状态更新为 0，主要是便于唤起后续节点，这个问题第四章节也已经聊过了，就不赘述了。
另外，当前节点的后继为 null 或者 后继节点的状态为 CANCEL，那么会从尾节点开始，从后往前寻找队列中最靠前的有效节点。
如果你觉得文章写的还不错，快给笔者点个赞吧，你的鼓励是笔者创作最大的支持！！！！！！
----------------------------
原文链接:https://www.jianshu.com/p/224e714b8548
程序猿的技术大观园：www.javathinker.net
[这个贴子最后由 flybird 在 2020-11-25 10:21:59 重新编辑]
|
|
Java技术网站 → 『 Java开发技术 』 → 『 Java编程技术专题 』 → 十分钟带你搞懂 Java AQS 核心设计与实现！
消息
查看
搜索
好友
邮件
复制
引用
IP: 已设置保密
