---
url: /java/AbstractQueuedSynchronizer.md
---
# 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
```

### 学习建议

1. **理解状态设计**：state 的含义由子类定义，是同步的核心
2. **掌握队列机制**：CLH 队列的入队、出队、唤醒流程
3. **区分两种模式**：独占和共享的差异在于唤醒传播
4. **阅读源码**：结合 ReentrantLock 等具体实现深入理解

***

## 参考资源

* JDK 源码：`java.util.concurrent.locks.AbstractQueuedSynchronizer`
* JSR-166 规范：https://jsr166.github.io/
* Doug Lea 论文：The java.util.concurrent Synchronization Framework
* 《Java 并发编程实战》第 14 章