AbstractQueuedSynchronizer (AQS) 原理及实现详解
一、概述
AbstractQueuedSynchronizer（简称 AQS）是 Java 并发包（JUC）的核心框架类，由 Doug Lea 编写，于 JDK 1.5 引入。它为实现阻塞锁和相关同步器（如信号量、事件等）提供了一个基于FIFO 等待队列的通用框架。
1.1 设计目的
AQS 的设计目标是：
- 为大多数同步器提供一个通用的基础框架
- 基于单个原子
int值表示同步状态 - 处理所有排队和阻塞机制，子类只需定义状态变化的保护方法
- 支持独占模式和共享模式两种同步方式
1.2 应用场景
AQS 是以下 JUC 组件的底层实现基础：
ReentrantLock（重入锁）ReentrantReadWriteLock（读写锁）CountDownLatch（倒计数器）Semaphore（信号量）Condition（条件变量）
二、核心数据结构
2.1 同步状态（state）
AQS 使用一个 volatile 的 int 字段来表示同步状态：
java
/**
* The synchronization state.
*/
private volatile int state;
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState;
}
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}状态的含义由子类定义：
- 在
ReentrantLock中：0 表示未锁定，1 表示已锁定，>1 表示重入次数 - 在
CountDownLatch中：表示剩余的计数值 - 在
Semaphore中：表示可用的许可数量
2.2 CLH 变体等待队列
AQS 使用一个双向链表结构的 FIFO 等待队列（CLH 队列的变体）：
头节点 (head) → 第一个等待节点 → ... → 尾节点 (tail)节点结构（Node 类）
java
abstract static class Node {
volatile Node prev; // 前驱节点
volatile Node next; // 后继节点
Thread waiter; // 等待的线程
volatile int status; // 节点状态
// 节点常量
static final int WAITING = 1; // 等待被唤醒
static final int CANCELLED = 0x80000000; // 已取消
static final int COND = 2; // 条件等待
}节点状态（status）
| 状态值 | 含义 | 说明 |
|---|---|---|
WAITING (1) | 等待中 | 节点需要被信号唤醒 |
CANCELLED (0x80000000) | 已取消 | 线程因超时或中断放弃获取锁 |
COND (2) | 条件等待 | 节点在条件队列中等待 |
2.3 队列示意图
+------+ prev +-------+ +------+
| head | <---- | first | <---- | tail |
+------+ +-------+ +------+
| | |
null thread1 thread3
↓
thread2三、两种同步模式
3.1 独占模式（Exclusive Mode）
- 特点：同一时刻只有一个线程能获取同步状态
- 典型应用：
ReentrantLock - 方法：
acquire(),release()
3.2 共享模式（Shared Mode）
- 特点：多个线程可同时获取同步状态
- 典型应用：
CountDownLatch,Semaphore,ReadWriteLock的读锁 - 方法：
acquireShared(),releaseShared()
四、核心方法
4.1 子类必须实现的方法
AQS 将同步逻辑委托给以下受保护的模板方法，子类必须重写：
| 方法 | 作用 | 返回值 |
|---|---|---|
tryAcquire(int arg) | 尝试独占获取 | true 成功 |
tryRelease(int arg) | 尝试独占释放 | true 完全释放 |
tryAcquireShared(int arg) | 尝试共享获取 | 负数失败，0 成功但后续可能失败，正数成功且后续也可能成功 |
tryReleaseShared(int arg) | 尝试共享释放 | true 可唤醒后续线程 |
isHeldExclusively() | 是否独占持有 | boolean |
4.2 AQS 提供的模板方法
这些方法是 final 的，不能被重写：
java
// 独占获取
public final void acquire(int arg)
public final void acquireInterruptibly(int arg) throws InterruptedException
public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException
// 独占释放
public final boolean release(int arg)
// 共享获取
public final void acquireShared(int arg)
public final void acquireSharedInterruptibly(int arg) throws InterruptedException
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException
// 共享释放
public final boolean releaseShared(int arg)五、核心流程分析
5.1 独占获取流程（acquire）
java
/**
* 独占获取的模板方法
* Acquires in exclusive mode, ignoring interrupts.
*/
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(null), arg))
reportInterruptAfterWait(null);
}流程图：
┌─────────────┐
│ tryAcquire │
└──────┬──────┘
│
┌────────────┴────────────┐
│ │
成功 │ 失败 │
▼ ▼
┌─────────────┐ ┌───────────────┐
│ 获取成功返回 │ │ addWaiter() │
└─────────────┘ │ 加入等待队列 │
└───────┬───────┘
│
▼
┌─────────────────┐
│ acquireQueued() │
│ 自旋尝试获取 │
└───────┬─────────┘
│
┌────────┴────────┐
│ │
成功 │ 失败 │
▼ ▼
┌──────────┐ ┌──────────┐
│ park() │◄─────│ 重试 │
│ 阻塞等待 │ └──────────┘
└────┬─────┘
│ 被唤醒
▼
┌──────────┐
│ 再次尝试 │
└──────────┘5.2 等待节点入队（addWaiter）
java
/**
* 将当前线程包装成 Node 并加入等待队列尾部
*/
private Node addWaiter(Node node) {
Node t = tail;
// 快速路径：队列已存在，直接 CAS 追加
if (t != null && u.compareAndSetReference(this, TAIL, t, node)) {
node.setPrevRelaxed(t);
return node;
}
// 慢速路径：初始化队列或处理竞争
for (int spins = 1; ; spins <<= 1) {
// ... 自旋等待后重新尝试
}
}5.3 队列中自旋获取（acquireQueued）
java
/**
* 在队列中自旋尝试获取同步状态
*/
final boolean acquireQueued(Node node, int arg) {
boolean failed = true;
try {
boolean wasInterrupted = false;
for (;;) {
Node p = node.getPrevious();
// 只有头节点的下一个节点才能尝试获取
if (p == head) {
if (tryAcquire(arg)) {
setHead(node); // 获取成功，设置为新头节点
return wasInterrupted;
}
}
// 检查是否需要阻塞
if (checkWaitNode(node, p))
LockSupport.park(this); // 阻塞当前线程
}
} finally {
if (failed) cancelAcquire(node);
}
}5.4 独占释放流程（release）
java
/**
* 独占释放的模板方法
*/
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.status != 0)
unparkSuccessor(h); // 唤醒后继节点
return true;
}
return false;
}
/**
* 唤醒后继节点
*/
private void unparkSuccessor(Node node) {
// 清除 WAITING 状态
if (node.getAndUnsetStatus(WAITING) >= 0)
return; // 无需唤醒
// 找到最前面的有效后继节点
Node s = node.getNext();
if (s == null || s.getStatus() <= 0) {
// 从尾部向前扫描找有效节点
for (Node t = tail; t != null && t != node; t = t.getPrevious()) {
if (t.getStatus() > 0) s = t;
}
}
if (s != null)
LockSupport.unpark(s.waiter); // 唤醒线程
}5.5 共享获取流程（acquireShared）
java
/**
* 共享获取的模板方法
*/
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg, null);
}
/**
* 共享模式的获取实现
*/
private void doAcquireShared(int arg, Long deadline) {
Node node = addWaiter(new Node());
boolean failed = true;
try {
for (;;) {
Node p = node.getPrevious();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
// 获取成功，传播给后继节点
doReleaseShared();
return;
}
}
if (checkWaitNode(node, p))
LockSupport.park(this);
}
} finally {
failed = false;
}
}5.6 共享释放的传播机制（doReleaseShared）
java
/**
* 共享释放时的传播操作
* 确保唤醒后继节点，即使当前节点不是最后一个
*/
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int hs = h.getStatus();
if (hs >= 0) {
// 设置 SIGNAL 状态并唤醒后继
if (h.compareAndSetStatus(hs, -hs))
unparkSuccessor(h);
}
else if (hs == 0)
// 设置 PROPAGATE 状态用于传播
h.compareAndSetStatus(0, -3); // PROPAGATE = -3
}
if (h == head) break;
}
}六、Condition 条件队列
6.1 ConditionObject 内部类
AQS 提供了 ConditionObject 作为 Condition 的实现：
java
public class ConditionObject implements Condition, Serializable {
private transient Object firstWaiter; // 条件队列头
private transient Object lastWaiter; // 条件队列尾
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 创建条件节点并加入条件队列
Node node = addConditionWaiter();
// 释放锁
int savedState = fullyRelease(node);
// 阻塞等待
while (!isSignalled())
LockSupport.park(this);
// 重新获取锁
acquire(savedState);
}
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
if (firstWaiter != null) {
doSignal(firstWaiter); // 唤醒第一个等待节点
}
}
}6.2 await/signal 流程
await() 流程:
1. 创建条件节点加入条件队列
2. 释放同步状态（解锁）
3. 阻塞等待（park）
4. 被唤醒后重新竞争锁
5. 恢复同步状态
signal() 流程:
1. 从条件队列移除节点
2. 将节点转移到主等待队列
3. 唤醒该节点（如果锁空闲）七、公平性与非公平性
7.1 默认策略：允许插队（Barging）
AQS 默认采用非公平策略，允许新来的线程"插队"：
java
// acquire 流程中，先尝试 tryAcquire，失败后才入队
if (!tryAcquire(arg) && acquireQueued(addWaiter(null), arg))这意味着：
- 新线程可能在入队前就获取到锁
- 已排队的线程被唤醒后可能需要重新竞争
- 优点：提高吞吐量，减少上下文切换
- 缺点：可能导致饥饿
7.2 实现公平锁
要实现公平锁，需要在 tryAcquire 中检查是否有前驱节点：
java
protected boolean tryAcquire(int acquires) {
if (!hasQueuedPredecessors() && // 检查是否有排队的前驱
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}八、实现示例
8.1 互斥锁（Mutex）
java
class Mutex implements Lock, Serializable {
private static class Sync extends AbstractQueuedSynchronizer {
// 尝试获取锁：state 从 0 变为 1
protected boolean tryAcquire(int acquires) {
assert acquires == 1;
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
// 尝试释放锁：state 从 1 变为 0
protected boolean tryRelease(int releases) {
assert releases == 1;
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
setExclusiveOwnerThread(null);
setState(0);
return true;
}
boolean isHeldExclusively() {
return getExclusiveOwnerThread() == Thread.currentThread();
}
Condition newCondition() {
return new ConditionObject();
}
}
private final Sync sync = new Sync();
public void lock() { sync.acquire(1); }
public boolean tryLock() { return sync.tryAcquire(1); }
public void unlock() { sync.release(1); }
public Condition newCondition() { return sync.newCondition(); }
public boolean isLocked() { return sync.getState() != 0; }
}8.2 布尔门闩（BooleanLatch）
java
class BooleanLatch {
private static class Sync extends AbstractQueuedSynchronizer {
boolean isSignalled() { return getState() != 0; }
// 共享获取：已 signalling 则成功 (返回 1)，否则失败 (返回 -1)
protected int tryAcquireShared(int ignore) {
return isSignalled() ? 1 : -1;
}
// 共享释放：设置状态为 1
protected boolean tryReleaseShared(int ignore) {
setState(1);
return true;
}
}
private final Sync sync = new Sync();
public boolean isSignalled() { return sync.isSignalled(); }
public void signal() { sync.releaseShared(1); }
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
}九、关键优化技术
9.1 CAS 无锁操作
AQS 大量使用 CAS（Compare-And-Swap）操作：
- 更新
tail指针 - 更新
state状态 - 更新节点链接
9.2 自旋与阻塞结合
- 在入队前先尝试获取（自旋）
- 入队后多次重试再阻塞
- 使用指数退避策略（最多 256 次重试）
9.3 垃圾回收优化
- 未入队的节点字段为
null - 出队的节点字段及时清空
- 减少内存占用
9.4 Dekker-like 信号方案
java
// 等待线程先设置 WAITING 状态
// 然后重试获取
// 再检查状态决定是否阻塞
// 释放线程清除 WAITING 状态后 unpark十、常见问题与注意事项
10.1 线程安全
- AQS 本身是线程安全的
- 子类实现的
tryXxx方法也必须是线程安全的 tryXxx方法应该简短且不阻塞
10.2 中断处理
acquireInterruptibly响应中断tryAcquireNanos支持超时和中断- 普通
acquire不响应中断
10.3 序列化
- AQS 只序列化
state字段 - 反序列化后队列为空
- 子类需要自定义
readObject方法
10.4 OOM 处理
- AQS 在无法分配节点时采用自旋等待
- Condition 等待遇到 OOM 时以固定速率重试
- 设计考虑了极端情况下的健壮性
十一、总结
AQS 的核心要点
| 方面 | 内容 |
|---|---|
| 核心思想 | 基于单个 int 状态 + FIFO 等待队列 |
| 两种模式 | 独占模式、共享模式 |
| 模板方法 | tryAcquire/tryRelease/tryAcquireShared/tryReleaseShared/isHeldExclusively |
| 队列类型 | CLH 变体双向链表 |
| 线程通信 | LockSupport.park/unpark |
| 公平性 | 默认非公平，可通过 hasQueuedPredecessors 实现公平 |
AQS 在 JUC 中的地位
AbstractQueuedSynchronizer
├── ReentrantLock.Sync
├── ReentrantReadWriteLock.Sync
│ ├── ReadLock.Sync
│ └── WriteLock.Sync
├── CountDownLatch.Sync
├── Semaphore.Sync
│ ├── FairSync
│ └── NonfairSync
└── SynchronousQueue.TransferStack学习建议
- 理解状态设计：state 的含义由子类定义，是同步的核心
- 掌握队列机制：CLH 队列的入队、出队、唤醒流程
- 区分两种模式：独占和共享的差异在于唤醒传播
- 阅读源码：结合 ReentrantLock 等具体实现深入理解
参考资源
- JDK 源码：
java.util.concurrent.locks.AbstractQueuedSynchronizer - JSR-166 规范：https://jsr166.github.io/
- Doug Lea 论文：The java.util.concurrent Synchronization Framework
- 《Java 并发编程实战》第 14 章