AbstractQueuedSynchronizer
AbstractQueuedSynchronizer (AQS) 源码分析
写一个锁,难点从来不在”改个标志位”,而在抢不到时的排队、阻塞、唤醒、取消这套脏活。AQS 把脏活做成模板:框架只管”一个 volatile int state + 一条 CLH 变体的 FIFO 队列”,子类只需用 state 定义”什么算抢到、什么算放掉”(tryAcquire / tryRelease),于是 ReentrantLock、Semaphore、CountDownLatch、ReentrantReadWriteLock 全在它上面长出来。
注意:JDK 15 起 AQS 被整个重写(JDK-8229442),老教程里的 waitStatus / SIGNAL / addWaiter / acquireQueued 在 JDK 17 源码里已经不存在,下面是重写后的真实代码。
// 基于 JDK 17, java.util.concurrent.locks.AbstractQueuedSynchronizer —— 骨架：状态位 + 节点 + state
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer // 只提供 exclusiveOwnerThread 字段，记录独占持有者
implements java.io.Serializable {
// ...
// Node status bits, also used as argument and return values
static final int WAITING = 1; // must be 1 // "我要睡了，释放时请叫我"，signalNext 只认这一位
static final int CANCELLED = 0x80000000; // must be negative // 符号位当取消标记，status < 0 一个比较就能判断
static final int COND = 2; // in a condition wait // 节点还在条件队列上，尚未转移到同步队列
/** CLH Nodes */
abstract static class Node {
volatile Node prev; // initially attached via casTail // 唯一可靠的链：casTail 成功即挂好
volatile Node next; // visibly nonnull when signallable // 只是加速用的"提示"，可能滞后甚至为 null
Thread waiter; // visibly nonnull when enqueued // 非 volatile：可见性搭 casTail/status 这些 volatile 写的顺风车
volatile int status; // written by owner, atomic bit ops by others // 自己普通写，别人用原子位运算改
// ... casPrev / casNext / getAndUnsetStatus / setPrevRelaxed 等 Unsafe 原子操作与字段偏移量省略
}
// Concrete classes tagged by type // 老版用 nextWaiter 字段区分模式，新版直接用类型：instanceof 判断更直白
static final class ExclusiveNode extends Node { }
static final class SharedNode extends Node { }
static final class ConditionNode extends Node
implements ForkJoinPool.ManagedBlocker { // 实现 ManagedBlocker：在 ForkJoinPool 里 await 不会耗死工作线程
ConditionNode nextWaiter; // link to next waiting node // 条件队列是单链表，不用 prev/next
// ...
}
/**
* Head of the wait queue, lazily initialized.
*/
private transient volatile Node head; // 哨兵节点，waiter 为 null；真正排队的从 head.next 开始
/**
* Tail of the wait queue. After initialization, modified only via casTail.
*/
private transient volatile Node tail; // 入队 = CAS 追加到 tail
/**
* @serial The synchronization state.
*/
private volatile int state; // 含义由子类定义：ReentrantLock 是重入次数，Semaphore 是剩余许可，CountDownLatch 是计数
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState; // 普通 volatile 写：只有持锁线程会调（如 tryRelease），无竞争不必 CAS
}
protected final boolean compareAndSetState(int expect, int update) {
return U.compareAndSetInt(this, STATE, expect, update); // 全家族无锁竞争的核心：抢锁本质是 CAS state
}
}// 基于 JDK 17, java.util.concurrent.locks.AbstractQueuedSynchronizer —— 独占获取：一个方法就是整台状态机
public abstract class AbstractQueuedSynchronizer {
// ReentrantLock.lock() 的入口：先无条件试一次（非公平插队点之一），失败才进大循环
public final void acquire(int arg) {
if (!tryAcquire(arg))
acquire(null, arg, false, false, false, 0L);
}
// 子类必须覆写；AQS 不关心"抢到"的语义，只管排队。默认抛异常而非 abstract：允许子类只实现独占或只实现共享
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
/**
* Main acquire method, invoked by all exported acquire methods.
* @return positive if acquired, 0 if timed out, negative if interrupted
*/
// 老版 addWaiter + acquireQueued + shouldParkAfterFailedAcquire + doAcquireShared 全部合并进这一个方法。
// 每轮循环只推进一小步（建队列 -> 建节点 -> 入队 -> 标记 WAITING -> park），走到哪一步由当前局部状态决定
final int acquire(Node node, int arg, boolean shared,
boolean interruptible, boolean timed, long time) {
Thread current = Thread.currentThread();
byte spins = 0, postSpins = 0; // retries upon unpark of first thread // 被唤醒后先自旋几次再睡，见下方"原理串讲"
boolean interrupted = false, first = false;
Node pred = null; // predecessor of node when enqueued
// ... 描述这台状态机十个分支的原文大段注释省略
for (;;) {
// 第一段：确认自己是不是"老二"（前驱就是 head）。只有老二有资格抢锁
if (!first && (pred = (node == null) ? null : node.prev) != null &&
!(first = (head == pred))) {
if (pred.status < 0) {
cleanQueue(); // predecessor cancelled // 前驱取消了，全队清理一遍再来
continue;
} else if (pred.prev == null) {
Thread.onSpinWait(); // ensure serialization // 前驱正在成为 head 的中间态，等它落定
continue;
}
}
// 第二段：老二 或 还没入队的新线程，都可以试抢 —— 后者就是非公平"插队"发生的地方
if (first || pred == null) {
boolean acquired;
try {
if (shared)
acquired = (tryAcquireShared(arg) >= 0);
else
acquired = tryAcquire(arg);
} catch (Throwable ex) {
cancelAcquire(node, interrupted, false); // 子类 tryAcquire 抛异常也要把节点摘干净，否则队列永久卡死
throw ex;
}
if (acquired) {
if (first) {
node.prev = null; // 抢到的节点自己变成新哨兵 head
head = node;
pred.next = null; // 旧 head 断链，帮 GC
node.waiter = null;
if (shared)
signalNextIfShared(node); // 共享模式的"传播"：我进来了，后面还是共享节点就接着叫醒
if (interrupted)
current.interrupt(); // 排队期间吃掉的中断在这里补回标志位
}
return 1;
}
}
// 第三段：抢失败，按状态推进一小步
Node t;
if ((t = tail) == null) { // initialize queue // 队列懒初始化：塞一个哨兵 ExclusiveNode
if (tryInitializeHead() == null)
return acquireOnOOME(shared, arg); // 连节点都 new 不出来（OOME）就退化成纯自旋，保证锁还能用
} else if (node == null) { // allocate; retry before enqueue
try {
node = (shared) ? new SharedNode() : new ExclusiveNode();
} catch (OutOfMemoryError oome) {
return acquireOnOOME(shared, arg);
}
} else if (pred == null) { // try to enqueue
node.waiter = current;
node.setPrevRelaxed(t); // avoid unnecessary fence // 先松弛写 prev，casTail 成功才算真入队
if (!casTail(t, node))
node.setPrevRelaxed(null); // back out
else
t.next = node; // next 在 CAS 之后才补，所以 next 天生可能滞后 —— prev 才是真相
} else if (first && spins != 0) {
--spins; // reduce unfairness on rewaits // 被唤醒又没抢过插队者：先自旋别急着再睡
Thread.onSpinWait();
} else if (node.status == 0) {
node.status = WAITING; // enable signal and recheck // 先竖"叫我"的牌子，再多循环一轮重试，防丢唤醒
} else {
spins = postSpins = (byte)((postSpins << 1) | 1); // 每被唤醒一次，下次自旋预算翻倍（上限 127）
try {
long nanos;
if (!timed)
LockSupport.park(this); // 真正挂起。blocker 传 this，jstack 里能看到在等哪个同步器
else if ((nanos = time - System.nanoTime()) > 0L)
LockSupport.parkNanos(this, nanos);
else
break; // 超时，掉到下面 cancelAcquire
} catch (Error | RuntimeException ex) {
cancelAcquire(node, interrupted, interruptible); // cancel & rethrow
throw ex;
}
node.clearStatus(); // 醒来先清 status，回到循环头重新抢
if ((interrupted |= Thread.interrupted()) && interruptible)
break; // 可中断模式：立刻放弃；不可中断模式：记下来继续抢
}
}
return cancelAcquire(node, interrupted, interruptible); // 只有中断/超时会走到这
}
// 取消 = 打上 CANCELLED 标记 + 触发 cleanQueue 摘链。返回值区分"报告中断"还是"吞掉并补标志"
private int cancelAcquire(Node node, boolean interrupted,
boolean interruptible) {
if (node != null) {
node.waiter = null;
node.status = CANCELLED;
if (node.prev != null)
cleanQueue();
}
if (interrupted) {
if (interruptible)
return CANCELLED;
else
Thread.currentThread().interrupt();
}
return 0;
}
}// 基于 JDK 17, java.util.concurrent.locks.AbstractQueuedSynchronizer —— 释放与唤醒：新版简单到出乎意料
public abstract class AbstractQueuedSynchronizer {
// ReentrantLock.unlock() 的入口。老版 unparkSuccessor 那套"从尾往前找"没了，就两行
public final boolean release(int arg) {
if (tryRelease(arg)) { // 子类判断是否真正释放（ReentrantLock：state 减到 0）
signalNext(head);
return true;
}
return false;
}
/**
* Wakes up the successor of given node, if one exists, and unsets its
* WAITING status to avoid park race. This may fail to wake up an
* eligible thread when one or more have been cancelled, but
* cancelAcquire ensures liveness.
*/
// 只叫 head.next 一个，且只在它竖了 WAITING 牌子时才叫 —— 没牌子说明它还在自旋，不需要 unpark。
// 允许漏叫（next 滞后/已取消时 s 可能不对），活性由 cleanQueue 里的补偿 signalNext 兜底
private static void signalNext(Node h) {
Node s;
if (h != null && (s = h.next) != null && s.status != 0) {
s.getAndUnsetStatus(WAITING); // 原子摘掉 WAITING 位再 unpark，同一个节点不会被重复叫醒
LockSupport.unpark(s.waiter);
}
}
// ========== 共享模式（Semaphore.acquire / CountDownLatch.await 走这里）==========
// 与独占共用同一台 acquire 状态机，只是 shared 传 true：成败判据变成 tryAcquireShared(arg) >= 0
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
acquire(null, arg, true, false, false, 0L);
}
// CountDownLatch.countDown / Semaphore.release 走这里
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
signalNext(head); // 只叫醒一个；"唤醒一串"靠 acquire 里的 signalNextIfShared 接力传播
return true;
}
return false;
}
/** Wakes up the given node if in shared mode */
// 传播的另一半：共享节点抢到并成为 head 后，发现下一个还是 SharedNode 就顺手叫醒它。
// 老版的 PROPAGATE 状态位 + doReleaseShared 循环，被这个"击鼓传花"取代
private static void signalNextIfShared(Node h) {
Node s;
if (h != null && (s = h.next) != null &&
(s instanceof SharedNode) && s.status != 0) {
s.getAndUnsetStatus(WAITING);
LockSupport.unpark(s.waiter);
}
}
// 公平锁的灵魂：队里有没有排在我前面的线程。FairSync.tryAcquire 抢锁前先问它
public final boolean hasQueuedPredecessors() {
Thread first = null; Node h, s;
if ((h = head) != null && ((s = h.next) == null ||
(first = s.waiter) == null ||
s.prev == null))
first = getFirstQueuedThread(); // retry via getFirstQueuedThread // next 只是提示，读不到就从 tail 沿 prev 重找
return first != null && first != Thread.currentThread(); // 队首是自己 = 没有前驱（重入/被叫醒的场景）
}
}原理串讲
拿默认的非公平 ReentrantLock 走一遍完整链路。线程 B 调 lock(),进 AQS.acquire(1),先无条件 tryAcquire(落到 NonfairSync 的 nonfairTryAcquire:CAS 把 state 从 0 改 1)。
锁被 A 持着,CAS 失败,进入六参数的 acquire(null, 1, false, false, false, 0L) 大循环。
B 的第一轮:node <mark> null 且 pred </mark> null,满足 first || pred == null,于是又 tryAcquire 一次——这是第二个插队点,B 此刻还没入队就能抢。
还是失败,往下走:tail == null 就先 tryInitializeHead 塞一个哨兵;下一轮 new ExclusiveNode();再下一轮 casTail 入队;再下一轮把 node.status 写成 WAITING;又转一轮确认还是抢不到,最后 LockSupport.park(this) 挂起。
注意这个节奏:每轮只推进一步,每步之间都夹着一次抢锁机会。
为什么 park 前要先写 status = WAITING、然后还要”多转一轮”才睡?这是无锁世界的经典丢唤醒问题:如果 B 检查完”抢不到”之后、park 之前,A 恰好 unlock 了,B 就会睡过去再也没人叫。
AQS 的解法是把”竖牌子”和”睡觉”拆成两轮——竖完 WAITING 后循环回头再抢一次,把这个窗口期的释放兜住;而 signalNext 那边只对 status != 0 的节点 unpark,配合 getAndUnsetStatus(WAITING) 原子摘牌,两边用一个 int 位就完成了握手,不需要任何锁。
就算真的先 unpark 后 park,LockSupport 的许可证语义也保证 park 立即返回。
A 这边 unlock() 进 release(1):tryRelease 把 state 减到 0、清掉 owner,然后 signalNext(head) 只叫醒 head.next 也就是 B。
为什么只叫一个?因为 head 后继以外的节点抢了也白抢(不是老二没资格),全叫醒就是惊群,n 个线程里 n-1 个白跑一趟再睡回去。
B 醒来后 clearStatus,回到循环头,head == pred 判定自己是老二,tryAcquire 成功后执行三连:node.prev = null; head = node; pred.next = null——B 的节点自己变成新哨兵,旧哨兵整个断链等 GC。
所以 head 永远是”当前持有者(或刚离开者)的座位”,队列里不存在单独的 dummy 对象常驻。
为什么 prev 是唯一可信的链,next 只是提示?入队的原子点是 casTail,而 setPrevRelaxed(t) 在 CAS 前、t.next = node 在 CAS 后——CAS 一成功 prev 链就完整,next 链却有一个真空期。
所以所有”必须正确”的操作(cleanQueue、getFirstQueuedThread、hasQueuedPredecessors 的兜底分支)都从 tail 沿 prev 反向走,next 只用来加速唤醒;signalNext 顺着 next 漏叫了也不要紧,cleanQueue 摘取消节点时会补一次 signalNext 保活性。
还有 spins/postSpins 这对不起眼的字段:非公平模式下 B 被叫醒后很可能被一个刚到的线程截胡(人家 CAS 一下就拿走了,B 还在从内核态爬起来)。
如果 B 每次被截胡都立刻重新 park,一次上下文切换几微秒就白烧了。所以 B 每被唤醒一次,postSpins 翻倍,下次先 Thread.onSpinWait() 自旋这么多次再考虑睡——被插队越狠,越倾向于原地等,把”非公平的吞吐优势”和”排队者不至于饿死太惨”折中在几行代码里。
设计取舍
- 模板方法的极致:AQS 只管排队/阻塞/唤醒,“什么算抢到”全交给子类的
tryAcquire/tryAcquireShared。同一套队列既能做锁(state=重入数)也能做信号量(state=许可数)、闭锁(state=计数),甚至读写锁把一个 int 拆成高 16 位读、低 16 位写。 tryAcquire默认抛UnsupportedOperationException而非声明 abstract:子类可以只实现独占(ReentrantLock)或只实现共享(Semaphore),不必为用不到的模式写空方法。- 非公平不是 bug 是默认:入队前有两次插队机会(
acquire(int)入口一次、大循环pred == null分支一次),换来的是省掉唤醒-接锁的切换延迟,吞吐显著更高;公平锁只是在tryAcquire里多问一句hasQueuedPredecessors。 - 版本大坑:JDK 15 重写后
waitStatus、SIGNAL(-1)、PROPAGATE(-3)、addWaiter、acquireQueued、unparkSuccessor全部不存在了,换成status位(WAITING/COND/CANCELLED)和一个合并的acquire状态机。
背旧版八股没问题,但别对着 JDK 17 的源码找 SIGNAL。 - 名字叫 CLH 变体,实际只借了”节点隐式排队”的思想:真 CLH 是在前驱的状态字段上自旋,AQS 抢不到直接 park 让出 CPU——自旋锁适合极短临界区,通用同步器必须睡。