Repository navigation
jdk concurrent aqs
wangkaiwu edited this page Apr 5, 2022
·
15 revisions
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
implements java.io.Serializable {
// 双向不循环链表
private transient volatile Node head;
private transient volatile Node tail;
private volatile int state;
}static final class Node {
static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;
// 标记当前节点是否失效
static final int CANCELLED = 1;
// 千万要注意，这个状态不是本节点在等待信号量，而是 next 节点在等待信号量！！！
static final int SIGNAL = -1;
static final int CONDITION = -2;
// 只作用于 head 节点，用于表示当前有可用的信号量
static final int PROPAGATE = -3;
// waitStatus == 0，也不是用来表示本节点的状态，而是表示后续节点无需被唤醒
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;
}public abstract class AbstractQueuedSynchronizer {
public final void acquire(int arg) {
// - 尝试请求 arg 个信号量
// - 添加到等待队列
// - 尝试获取信号量，获取不到就会 park，之后等待被唤醒。
// - 无论是什么原因被唤醒，都会重试获取信号量。
// - 如果获取锁之前曾经被中断过，则进行
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
}public abstract class AbstractQueuedSynchronizer {
// 请注意该方法唯一的 return 位置。
// 只有原子的将自己放到队尾才会返回。
// 该方法存在一个竟态条件 tail 的值
private Node addWaiter(Node mode) {
Node node = new Node(mode);
for (; ; ) {
// 尾节点
Node oldTail = tail;
if (oldTail != null) {
// 等待队列已经初始化过了
// 等同于 node.setPrev(oldTail)，之所以不直接set，因为 prev 变量是 volatile
// 由于下一行指令本身就涉及编译器和内存双屏障，所以这里就相当于省掉了一次屏障.
// ps: 换成 AtomicReference 就自带 lazySet 方法了
node.setPrevRelaxed(oldTail);
// cas 地将自己作为 tail
if (compareAndSetTail(oldTail, node)) {
// 既然自己成了 tail，就把自己加到 旧的 tail 后面
oldTail.next = node;
return node;
}
} else {
// 初始化
initializeSyncQueue();
}
}
}
// 所谓的初始化，就是将 head 和 tail 设置为空节点。
// 也可以认为head 和 tail 各有一个哨兵。
// 哨兵无法 park 和 unpark，因为有关联线程。
// 哨兵初始 waitStatus 是 0
private final void initializeSyncQueue() {
Node h;
if (HEAD.compareAndSet(this, null, (h = new Node())))
tail = h;
}
}public abstract class AbstractQueuedSynchronizer {
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
for (; ; ) {
// 等同于 p = node.prev();
final Node p = node.predecessor();
// 时刻要注意一点，head 有可能是哨兵。
// 如果 head 不是哨兵，则一定是一个获取到信号量的节点。
// 判定 p==head 不会存在多线程问题，理由是只有一个节点的 prev 会是 head 节点。
// 因为从后往前遍历不会存在多线程问题, addWaiter 方法可以保证这一点
if (p == head && tryAcquire(arg)) {
// 如果信号量获取成功，则完全可以将自己作为头节点。
// - head 如果是哨兵，替换掉哨兵无可厚非
// - head 如果是其他获取到信号量的线程。它都已经获取到信号了，也可以被替换掉了。
setHead(node);
// 旧的 head 节点已经没用了。
p.next = null; // help GC
return interrupted;
}
// 走到这里有两种可能：
// 1. 还没轮到自己获取信号量，node.prev 不是 head
// 2. 信号量不够用。tryAcquire 失败
if (shouldParkAfterFailedAcquire(p, node)) {
// 如果 p 的状态恰好是 SIGNAL 则执行
interrupted |= parkAndCheckInterrupt();
}
// 只要 prev 不是 SIGNAL 状态，都会重试
}
} catch (Throwable t) {
cancelAcquire(node);
if (interrupted)
selfInterrupt();
throw t;
}
}
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
// 检查并更新未能获取的节点的状态。
// 如果线程应该阻塞，则返回 true。
// 这是所有采集循环中的主要信号控制。
// 要求 pred == node.prev。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) {
// 由于 pred.waitStatus 标记的是下一个节点的状态，
// 说明 node 已经处于了等待获取信号量状态
// 可以直接 park
return true;
}
// CANCELLED = 1. 只有取消状态 > 0
// CANCELED 表示的是节点自身的状态，说明这个节点需要移除
if (ws > 0) {
// 说明 pred 应该要从等待队列移除
do {
// A -> B -> C。 B 是 pred， C 是node
// 执行后， A -> C。 也就是把 B 移除了。
// 由于只修改了 node.prev，所以 node.next 其实还是旧值
// 换句话说，想要遍历必须从后往前开始, 从 tail -> head。绝不可以 head -> tail
node.prev = pred = pred.prev;
// 现在的 pred 是 A。意思是看下 A 是否也需要移除
// 但是 pred 的尽头是 head，而 head 有两种可能
// - 哨兵：哨兵的状态永远是 <= 0，不可能处于 CANCELD 状态，没有意义。
// - 已经获取到信号量的节点：都已经获取到信号量了，在设置为 CANCELED 还有什么意义呢。
// 综上因为哨兵的存在，所以最后一个 pred 的值肯定 <=0。精髓啊 ！！！
} while (pred.waitStatus > 0);
// do-while 的作用就是移除失效节点，
// 跳出循环的 pred 是离 node 最近的一个有效节点。
// 通过下面的方法完成对失效节点的彻底删除
pred.next = node;
// 既然前面有节点被移除了，可以再尝试一下是否能成功获取信号量
} else {
// 如果 pred 当前状态 == 0，而自己就是因为获取不到信号量才来到的这里
// 所以需要把前一个节点的 waitStatus 改为 SIGNAL，表示自己在等待获取信号
// 是的，SIGNAL 的作用是用来标记下一个节点需要信号量
pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
// 还会再尝试获取一次信号量
}
return false;
}
private final boolean parkAndCheckInterrupt() {
// 阻塞线程
LockSupport.park(this);
// 被唤醒了。可能是被 release 唤醒，也可能是被 interrupt 唤醒
return Thread.interrupted();
}
}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
implements java.io.Serializable {
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
// 友情提醒：head 可能是哨兵，也可能是 release 的调用者。。。。
// 这里的条件很有意思:
// h != null: 只要没有发生竟态，head 就会一直为空。从侧面来讲，也不存在等待者
// h.waitStatus != 0: 由此可知，waitStatus == 0 表示后续节点无需被唤醒
// head.waitStatus == 0: 表示的是等待队列的节点都无需唤醒，直白点儿就是没人在等待
if (h != null && h.waitStatus != 0) {
unparkSuccessor(h);
}
// 如果 h 已经获取到了信号量，就不管了。反正 h 早晚也会释放信号量
return true;
}
return false;
}
}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
implements java.io.Serializable {
private void unparkSuccessor(Node node) {
// 期望 node 的状态是 SIGNAL(-1)
int ws = node.waitStatus;
if (ws < 0) {
// ws 可能处于 SIGNAL(-1) PROPAGATE(-3) 状态
// SIGNAL: 由于后面的逻辑就是为了唤醒后续节点，
// 所以提前将状态置为后续节点无需唤醒，也就是0
// PROPAGATE: 无条件传播状态 !!!
node.compareAndSetWaitStatus(ws, 0);
}
// 这里探讨 node 是谁的问题
// - 是本人：既然是唤醒者本人，唤醒 next 也就是下一个等待者无可厚非。
// - 是哨兵：哨兵不管，唤醒的只能是 next，哨兵没必要唤醒。
// - 是他人：他人必然也拿到信号量了，否则不会到 head 位置上来.
// 所以也没有必要唤醒,所以更加应该唤醒 node.next
// 后续节点
Node s = node.next;
// CANCELED(1) s 可能被 cancel 了。
// 准备唤醒下一个等待者，结果这个人不存在，或者状态是 CANCELED(1)
if (s == null || s.waitStatus > 0) {
// help gc
s = null;
// 逐步分析, p 是尾部
// p != node && p != null 这可以转化成 p != null && p != node
// p 如果等于空，就表示队列是空的，因为 p 是队尾。
// p 如果等于 node，而 node 是 head，也就是判断一下队列是否只有 p 这一个节点
// p 的 waitStatus <= 0，说明这个节点的状态正常
// p = p.prev ： p 是队尾，p.prev 是队尾的前一个节点。
// 翻译一下：就是从队尾开始往前遍历，找一个最靠前的状态正常的节点，然后将其唤醒
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
// 只要节点存在就唤醒
if (s != null)
LockSupport.unpark(s.thread);
}
}count = 1，A 请求 1 信号量, B 请求 1 信号量
| state | count | head | tail | queue | 备注 |
|---|---|---|---|---|---|
| init | 1 | null | null | ||
| A acquire 1 | 0 | null | null | ||
| B acquire 1 | 0 | None | None | None(0) <-> None(0) | |
| B park | 0 | None | B | None(S)<->B(S) | |
| A release 1 | 1 | None | B | None(S)<->B(S) | |
| A unpark B | 1 | None | B | None(S)<->B(0) | B 的 next 还是 B，所以 A 唤醒 B |
| B acquire 1 | 0 | B | B | B(0)<-> B(0) ; None(0)-> null | |
| B release 1 | 1 | B | B | B(0)<- B(0) ; B(0)-> null |
public abstract class AbstractQueuedSynchronizer {
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0) {
// 信号量不足
doAcquireShared(arg);
}
// 获取成功
}
private void doAcquireShared(int arg) {
// SHARED = new Node();
// node.nextWaiter = Node.SHARED;
final Node node = addWaiter(Node.SHARED);
boolean interrupted = false;
try {
for (; ; ) {
// 上一个节点
final Node p = node.predecessor();
// head 可能是哨兵，也可能是已获得信号量的节点
if (p == head) {
// r 是扣除 arg 后剩余信号量
int r = tryAcquireShared(arg);
if (r >= 0) {
// 信号量获取成功, 将自己作为 head 节点
// 这里非常像是循环唤醒。以 None -> A -> B -> C 为例。
// A 被唤醒，获取了信号量，那么就会调用唤醒函数把 head.next 唤醒，也就是 B。
// 如果 B 被A唤醒了，会以同理唤醒 C。直到无法唤醒为止。
setHeadAndPropagate(node, r);
p.next = null; // help GC
return;
}
}
// 只要自己不是 head —> next 节点，就会直接 park
if (shouldParkAfterFailedAcquire(p, node)) {
// 如果头节点是哨兵，而自己有没有获得信号量，所以 park
// 如果前一个不是头结点，也需要 park.
// 因为会有 head + 1 节点来保证吧自己唤醒
interrupted |= parkAndCheckInterrupt();
}
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
} finally {
if (interrupted)
selfInterrupt();
}
}
}public abstract class AbstractQueuedSynchronizer {
// 此方法是获取信号量成功后调用，
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
// 按照惯例，获取信号量成功后将自己作为头节点
// head 原本是哨兵：既然是哨兵，就需要顶掉哨兵，自己当头节点
// head 是获取了信号量的节点：既然前一个 head 已经拿到信号量了，也会顶掉，因为不再用了
setHead(node);
if (propagate > 0 || // 还有空闲的信号量
h == null || // 没有可用信号量，但是原本的头节点也不存在。自己成为头节点之前不存在头结点
h.waitStatus < 0 || // 原本的头结点也存在。但是 h 的 next 在等待信号量
(h = head) == null || // 原本的头节点显示没有人在等待信号，那就再拿一下最新的头结点(我认为此时为空的可能性极小)
h.waitStatus < 0) { // 最新的头结点的状态显示，后续节点在等待信号量，或者有可用信号量
Node s = node.next;
// 因为当前是共享资源模式，所以允许同时有多个线程获取信号量
if (s == null || s.isShared()) {
// 如果 node 后面没有节点了，或者是共享节点，就尝试唤醒一下
// 为何 s == null 还需要唤醒???
doReleaseShared();
}
}
}
}public abstract class AbstractQueuedSynchronizer {
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
// 信号量释放成功
doReleaseShared();
return true;
}
return false;
}
private void doReleaseShared() {
for (; ; ) {
Node h = head;
if (h != null && h != tail) {
// 只要队列中不止一个元素就进来
int ws = h.waitStatus;
// ws == SIGNAL，表示后续节点在等待信号量
if (ws == Node.SIGNAL) {
// 后面就会唤醒后续节点了，所以先把状态改成后续节点无需被唤醒
if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0)) {
// 没能把 head 的状态改回来
continue; // loop to recheck cases
}
// unpark 的主要作用就是唤醒 h.next
unparkSuccessor(h);
} else if (ws == 0 &&
!h.compareAndSetWaitStatus(0, Node.PROPAGATE)) {
// ws == 0: 表示后续节点无需唤醒
// 将状态改为 PROPAGATE，主要是服务于 setHeadAndPropagate
// 对于 doReleaseShared 方法而言，有没有 PROPAGATE 状态无区别
continue; // loop on failed CAS
}
// 走到这里，要么执行了 if 中的 unpark 唤醒了后续节点
// 要么是在 else if 中把状态改为了 PROPAGATE
}
// 如果 head 变化就再次循环
if (h == head) // loop if head changed
break;
}
}
}设想一个场景，release 的时候发现等待队列至少有一个等待者。 而 head.next 这个等待者却没有将 head 的状态置为 SIGNAL。 这有两种可能:
- 等待者还没来得及将状态设为 SIGNAL。这个情况没有关系，因为无论如何都会重试。
- 刚释放完信号量等待者(head.next)已经拿到了信号量。
public abstract class AbstractQueuedSynchronizer {
private final boolean compareAndSetTail(Node expect, Node update) {
// 等同于 oldTail = tail;
// if tail == expect;
// tail = update;
// return oldTail;
return TAIL.compareAndSet(this, expect, update);
}
}static final class Node {
volatile int waitStatus;
final Node predecessor() {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
final boolean compareAndSetWaitStatus(int expect, int update) {
// 等同于 waitStatus.cas(expect, update);
return WAITSTATUS.compareAndSet(this, expect, update);
}
}public class LockSupport {
private static final Unsafe U = Unsafe.getUnsafe();
private static final long PARKBLOCKER = U.objectFieldOffset(Thread.class, "parkBlocker");
private static void setBlocker(Thread t, Object arg) {
// 等价于 t.setParkBlocker(arg);
// 由于 Thread 类的 parkBlocker 成员是 volatile 修饰的，然儿这里不想让编译器产生一个写屏障影响性能,所以就用了 unsafe 的方式调用。
// volatile Object parkBlocker;
U.putObject(t, PARKBLOCKER, arg);
}
// blocker 对应的是 AQS 对象本身。方法的作用是让当前线程休眠
public static void park(Object blocker) {
Thread t = Thread.currentThread();
// == t.setParkBlocker(blocker); blocker 的主要作用是给 jfr 提供数据。
setBlocker(t, blocker);
// false 表示后面的时间不是绝对值，0L 表示不超时
U.park(false, 0L);
setBlocker(t, null);
}
}public
class Thread implements Runnable {
public static boolean interrupted() {
return currentThread().isInterrupted(true);
}
private native boolean isInterrupted(boolean ClearInterrupted);
}public class Thread implements Runnable {
private volatile Interruptible blocker;
private final Object blockerLock = new Object();
public void interrupt() {
if (this != Thread.currentThread()) {
checkAccess();
synchronized (blockerLock) {
// 目前看不出来在哪里赋值的
Interruptible b = blocker;
if (b != null) {
interrupt0(); // set interrupt status
b.interrupt(this);
return;
}
}
}
// 唤醒线程的一切 park() 状态
interrupt0();
}
private native void interrupt0();
}