AbstractQueuedSynchronizer (AQS) 源码精读
概述
AbstractQueuedSynchronizer（AQS）是 java.util.concurrent 的基石：ReentrantLock、Semaphore、CountDownLatch、ReentrantReadWriteLock、ThreadPoolExecutor.Worker 全部构建在它之上。AQS 用一个 volatile int state + 一条 CLH 变体等待队列实现同步器的通用框架，子类只需实现 tryAcquire / tryRelease（独占）或 tryAcquireShared / tryReleaseShared（共享）四个模板钩子。
本文基于 OpenJDK 21 源码，拆解 AQS 的队列结构、获取/释放流程、条件队列与底层原子操作。
一、核心属性与设计
1.1 state 核心属性
java
// java.util.concurrent.locks.AbstractQueuedSynchronizer
private volatile int state;state 是同步状态，含义由子类定义：ReentrantLock 用它记录重入次数，Semaphore 用它记录剩余许可数，CountDownLatch 用它记录剩余计数。
java
protected final int getState() { return state; }
protected final void setState(int newState) { state = newState; }
// 原子 CAS 更新（底层走 VarHandle，见第八节）
protected final boolean compareAndSetState(int expect, int update) {
return STATE.compareAndSet(this, expect, update);
}
getState()/setState()是普通读写（用于持有锁后更新），compareAndSetState()是 CAS 原子更新（用于竞争获取）。
1.2 Node 的 CLH 变体队列
java
// AQS 内部类 Node
static final class Node {
static final Node SHARED = new Node(); // 共享模式标记
static final Node EXCLUSIVE = null; // 独占模式标记
static final int CANCELLED = 1; // 节点已取消（只有此值为正）
static final int SIGNAL = -1; // 后继节点等待唤醒
static final int CONDITION = -2; // 节点在条件队列中
static final int PROPAGATE = -3; // 共享模式传播释放
volatile int waitStatus; // 等待状态
volatile Node prev; // 前驱（CLH 需要前驱）
volatile Node next; // 后继
volatile Thread thread; // 排队线程
Node nextWaiter; // 条件队列后继 / SHARED 标记
}与经典 CLH 锁不同，AQS 的变体：节点记录等待线程（用于 LockSupport.park / unpark），waitStatus 标记状态（SIGNAL 表示"我释放时唤醒后继"），是"显式前驱通知"而非自旋轮询。
二、独占锁获取：acquire(int)
2.1 模板方法
java
// AQS 独占获取（模板方法，final）
public final void acquire(int arg) {
if (!tryAcquire(arg) && // ① 子类钩子：尝试获取
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // ② 入队并自旋
selfInterrupt(); // ③ 中断被挂起时补上中断标记
}流程：先试一次 tryAcquire（非公平锁的抢占机会），失败则 addWaiter 入队，再 acquireQueued 自旋等待，直到拿到锁或被打断。
java
// 子类实现的钩子示例（ReentrantLock.NonfairSync）
protected final boolean tryAcquire(int acquires) {
// CAS 抢锁 → 成功则记录独占线程并返回 true
}2.2 入队：addWaiter(Node)
java
// 快速路径：队列非空 → CAS 尾插
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) { // CAS 更新尾节点
pred.next = node;
return node;
}
}
enq(node); // 慢路径：队列为空或 CAS 失败 → 自旋
return node;
}
// 自旋初始化 + 尾插
private Node enq(final Node node) {
for (;;) {
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;
}
}
}
}
enq的自旋保证多线程并发入队只有一个成功；head是哨兵节点，持有锁的线程对应head，不参与排队。
2.3 自旋等待：acquireQueued(Node, int)
java
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
for (;;) {
final Node p = node.predecessor(); // 前驱
if (p == head && tryAcquire(arg)) { // 前驱是头 → 再试一次
setHead(node); // 获取成功 → 自己成为头
p.next = null; // 旧头出队（帮助 GC）
return interrupted;
}
// 获取失败 → 决定是否 park 并处理中断
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true; // 记录中断，不立即抛出
}
} catch (Throwable t) {
cancelAcquire(node); // 异常 → 取消节点
throw t;
}
}要点：只有前驱是 head 的节点才有资格尝试获取，保证 FIFO 大体公平；中断不直接抛出，而是记录后在 acquire 末尾补 selfInterrupt()。
三、shouldParkAfterFailedAcquire 的 waitStatus 传播
java
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) // 前驱已承诺唤醒我 → 可以安心 park
return true;
if (ws > 0) { // CANCELLED → 跳过所有取消节点
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else { // 0 或 PROPAGATE → CAS 设为 SIGNAL
pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
}
return false; // 本次不 park，外层循环重试
}状态传播逻辑：每个节点在 park 前必须保证前驱的 waitStatus 被置为 SIGNAL——这样前驱释放锁时会 unpark 后继，避免"先 park 后错过唤醒"的竞态。
四、独占锁释放：release(int)
java
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) {
int ws = node.waitStatus;
if (ws < 0) node.compareAndSetWaitStatus(ws, 0); // 清除 SIGNAL
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); // 唤醒后继线程
}唤醒时从尾向头查找的原因：
next指针的建立晚于prev（先pred.next = node可能在并发下指向已取消节点），prev链是可靠的。
五、ConditionObject 条件队列
5.1 await() 等待
java
public final void await() throws InterruptedException {
if (Thread.interrupted()) throw new InterruptedException();
Node node = addConditionWaiter(); // ① 包装成 CONDITION 节点入条件队列
int savedState = fullyRelease(node); // ② 释放锁（可重入时全量释放）
int interruptMode = 0;
while (!isOnSyncQueue(node)) { // ③ 不在同步队列 → park
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// ④ 被 signal 后进入同步队列，重新抢锁（走 acquireQueued）
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// ⑤ 清理取消节点 / 补中断
if (node.nextWaiter != null) unlinkCancelledWaiters();
if (interruptMode != 0) reportInterruptAfterWait(interruptMode);
}fullyRelease 是关键：await 必须完全释放锁（包括重入次数），否则别的线程进不来，signal 也无法唤醒。
java
final int fullyRelease(Node node) {
int savedState = getState();
if (release(savedState)) return savedState; // 一次性释放全部重入计数
// 失败 → 节点标记取消并抛异常
}5.2 signal() 通知
java
public final void signal() {
if (!isHeldExclusively()) throw new IllegalMonitorStateException();
Node first = firstWaiter; // 队首等待者
if (first != null) doSignal(first);
}
private void doSignal(Node first) {
do {
if ((firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) && // 转移到同步队列失败（取消）则继续
(first = firstWaiter) != null);
}
final boolean transferForSignal(Node node) {
if (!node.compareAndSetWaitStatus(Node.CONDITION, 0)) return false;
Node p = enq(node); // 条件队列 → 同步队列尾部
int ws = p.waitStatus;
// 前驱已取消或 CAS 置 SIGNAL 失败 → 直接唤醒该线程
if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}要点：signal 只是把条件队列头节点转移到同步队列（设置 SIGNAL 由前驱负责唤醒），并不保证立即 unpark——唤醒交给前驱释放锁时完成，除非前驱已取消。
六、VarHandle 在 AQS 中的应用
java
// java.util.concurrent.locks.AbstractQueuedSynchronizer
private static final VarHandle STATE;
private static final VarHandle HEAD;
private static final VarHandle TAIL;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
STATE = l.findVarHandle(AbstractQueuedSynchronizer.class, "state", int.class);
HEAD = l.findVarHandle(AbstractQueuedSynchronizer.class, "head", Node.class);
TAIL = l.findVarHandle(AbstractQueuedSynchronizer.class, "tail", Node.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
}AQS 在 JDK 9 后用 VarHandle 替代了 Unsafe 的 CAS 操作：
java
protected final boolean compareAndSetState(int expect, int update) {
return STATE.compareAndSet(this, expect, update);
}VarHandle 相比 Unsafe 的优势：类型安全（编译期检查）、可在任何访问模式下工作（包括不可变对象）、受 JMM 正式规范约束、可移植到 GraalVM 等非 HotSpot 实现。
七、实现要点
AQS 源码核心：
state 属性：volatile 同步状态，子类定义语义
CLH 变体队列：prev/next/thread/waitStatus，前驱 SIGNAL 通知
acquire：tryAcquire → addWaiter → acquireQueued（自旋 + park）
release：tryRelease → unparkSuccessor（从尾找后继唤醒）
Condition：await 全量释放 + park，signal 转移节点
VarHandle：类型安全的 CAS 原子操作
常见陷阱：
tryAcquire 失败不抛异常 → 需要返回 false 入队
await 未完全释放重入锁 → 死锁
signal 后仍要重新抢锁（不是立即获得）
中断在 acquire 中被记录而非立即抛出与其他同步器联动
AQS 与上层同步器：
ReentrantLock → tryAcquire/tryRelease（重入计数）
Semaphore → tryAcquireShared/tryReleaseShared（许可数）
CountDownLatch → tryAcquireShared（计数为 0 放行）
ReentrantReadWriteLock → 读共享 + 写独占（高低位拆分）
ThreadPoolExecutor.Worker → 自身 AQS（不可重入独占）