前言

在 Java 并发编程中，锁和同步器是不可或缺的工具。从 synchronized 到 ReentrantLock，从 CountDownLatch 到 Semaphore，这些工具的背后都有一个共同的基石——AbstractQueuedSynchronizer（AQS）。

AQS 是 java.util.concurrent 包的灵魂，它提供了一套框架，用于实现依赖 FIFO 等待队列的阻塞锁和同步器。JDK 中大部分的同步工具都是基于 AQS 构建的。理解 AQS 的运行机制，对于理解 Java 并发编程有着很大的帮助。

核心设计思想

AQS 的核心可以概括为两个部分：

  • state：一个 volatile 修饰的 int 变量，表示同步状态
  • CLH 队列：一个 FIFO 的双向队列，用于存储等待获取锁的线程

state 变量

AQS 通过 volatile int state 来表示同步状态，子类通过继承 AQS 并实现其模板方法来管理 state 的变化。以 ReentrantLock 为例：

  • state = 0 表示锁没有被持有
  • state = 1 表示锁被某个线程持有
  • state > 1 表示锁被同一个线程重入了 N 次

state 的更新使用 compareAndSetState 方法，利用 Unsafe 的 CAS 操作保证原子性。

1
2
3
4
5
private volatile int state;

protected final boolean compareAndSetState(int expect, int update) {
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

CLH 等待队列

CLH 队列本质上是一个双向链表，AQS 将每个获取锁失败的线程封装成一个 Node 节点，并插入到队列尾部。

1
2
3
4
5
6
7
static final class Node {
    volatile int waitStatus;    // 等待状态
    volatile Node prev;         // 前驱节点
    volatile Node next;         // 后继节点
    volatile Thread thread;     // 持有该节点的线程
    Node nextWaiter;            // 指向下一个等待 condition 的节点
}

队列的结构可以用下图来理解：

graph LR
    head["head (哨兵节点)"] --> node1["Node1
thread=A
waitStatus=0"]
    node1 --> node2["Node2
thread=B
waitStatus=0"]
    node2 --> tail["tail"]
    tail -.-> node2
    node2 -.-> node1
    node1 -.-> head
    style head fill:#f9f,stroke:#333
    style tail fill:#f9f,stroke:#333

head 节点是一个哨兵节点（不关联具体线程），每个等待的线程都有自己的 Node 节点，prev 和 next 指针构成了双向链表。

独占模式下的获取与释放

AQS 提供了 acquire 和 release 作为独占模式下获取和释放同步状态的入口。

acquire - 获取同步状态

1
2
3
4
5
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

这个方法的执行流程如下：

  1. 调用 tryAcquire 尝试获取同步状态（由子类实现）
  2. 如果失败，调用 addWaiter 将当前线程包装成 Node 节点加入等待队列
  3. 调用 acquireQueued 让节点在队列中自旋等待
  4. 如果线程被中断，响应中断

整个过程可以看作：

flowchart TD
    A[调用 acquire] --> B{tryAcquire
获取同步状态?}
    B -->|成功| C[执行业务逻辑]
    B -->|失败| D[addWaiter
构建Node加入队尾]
    D --> E[acquireQueued
自旋等待]
    E --> F{前驱节点
是 head?}
    F -->|否| G[阻塞当前线程
等待前驱唤醒]
    G --> E
    F -->|是| H[再次 tryAcquire]
    H -->|成功| I[设置为新 head
返回中断状态]
    H -->|失败| G
    I --> C

addWaiter - 入队

1
2
3
4
5
6
7
8
9
10
11
12
13
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)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);  // CAS 失败或队列未初始化，走自旋入队
    return node;
}

代码先尝试一次快速的 CAS 入队，如果失败则调用 enq 方法进行自旋入队，保证线程安全。

acquireQueued - 自旋等待

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null;
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

核心逻辑很简单：只有当前驱节点是 head 时才会尝试获取锁，获取成功则成为新的 head。如果获取失败，调用 shouldParkAfterFailedAcquire 检查是否应该挂起线程，如果前驱节点的 waitStatus 被设置为 SIGNAL（表示它会在释放锁时唤醒后继），则当前线程可以安全挂起。

release - 释放同步状态

1
2
3
4
5
6
7
8
9
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

释放成功后会调用 unparkSuccessor 唤醒 head 的下一个等待线程。释放流程相对简单，主要是因为复杂的排队逻辑已经在 acquire 中处理了。

flowchart LR
    A[持有锁的线程执行完毕] --> B[release]
    B --> C{tryRelease
释放成功?}
    C -->|否| D[直接返回 false]
    C -->|是| E{head 不为 null
且 waitStatus != 0?}
    E -->|否| F[返回 true
无等待线程]
    E -->|是| G[unparkSuccessor
唤醒后继线程]
    G --> H[后继线程继续 acquireQueued]

共享模式下的获取与释放

共享模式与独占模式的核心区别在于：共享模式下，当一个线程获取到同步状态后，可以同时通知后续的等待线程。

acquireShared

1
2
3
4
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

tryAcquireShared 返回 int 值：

  • 正数：获取成功，且后续线程也可以尝试获取
  • 零：获取成功，但后续线程不能获取
  • 负数：获取失败

独占 vs 共享对比

特性 独占模式 共享模式
同一时刻 只有一个线程持有 多个线程可以同时持有
典型实现 ReentrantLock Semaphore、CountDownLatch
唤醒策略 只唤醒下一个节点 传播唤醒后续节点
tryXxx 返回值 boolean int（正/零/负）

基于 AQS 的实现

ReentrantLock

ReentrantLock 是 AQS 独占模式最典型的应用。它的核心是内部类 Sync 继承 AQS，并实现了 tryAcquire 和 tryRelease。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 非公平锁的 tryAcquire 实现
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // state 为 0，直接 CAS 尝试获取
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) {
        // 可重入逻辑
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

这里体现了两个关键点：

  • 非公平性：新来的线程可以和队列中的线程竞争（先 CAS 尝试再入队）
  • 可重入：同一个线程可以多次获取锁，state 递增累计

Semaphore

Semaphore（信号量）基于 AQS 的共享模式实现。它通过 permits（许可证数量）来限制同时访问资源的线程数。

1
2
3
4
5
6
7
8
9
10
// 非公平信号量的 tryAcquireShared
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

当 state 代表的剩余许可证不够时，返回负数，线程进入等待队列。

CountDownLatch

CountDownLatch 基于 AQS 共享模式实现，使用 state 代表计数器的初始值。countDown() 调用 releaseShared 递减 state；await() 调用 acquireShared 阻塞直到 state 为 0。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 内部 Sync 的 tryAcquireShared
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

// tryReleaseShared
protected boolean tryReleaseShared(int releases) {
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        int nextc = c - 1;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

state 减到 0 时，tryReleaseShared 返回 true，AQS 会唤醒所有在 await 上等待的线程。

三种同步器对比

flowchart TD
    subgraph AQS核心
        state["state (volatile int)"]
        queue["CLH FIFO 双向队列"]
    end

    subgraph 独占模式
        ReentrantLock["ReentrantLock
state=0 无锁
state=1 有锁
state>1 重入次数"]
    end

    subgraph 共享模式
        Semaphore["Semaphore
state=剩余许可证数"]
        CountDownLatch["CountDownLatch
state=初始计数"]
    end

    AQS核心 --> ReentrantLock
    AQS核心 --> Semaphore
    AQS核心 --> CountDownLatch

总结

AQS 的设计体现了几个重要的编程理念：

  1. 模板方法模式：AQS 定义了获取/释放同步状态的骨架，子类只需实现 tryAcquire、tryRelease、tryAcquireShared、tryReleaseShared、isHeldExclusively 等几个方法。
  2. 自旋 + CAS：入队和出队操作通过自旋 + CAS 保证线程安全，避免使用重量级锁。
  3. LockSupport：线程的阻塞和唤醒依赖 LockSupport.park() 和 unpark()，比 wait/notify 更灵活。
  4. 可扩展性：基于 AQS 可以方便地实现自定义同步器，如手写一个一次性门阀（类似于只有一次计数的 CountDownLatch）。

理解 AQS，本质上就理解了 Java 并发包中半数以上同步工具的实现原理，这也是为什么我们说 AQS 是 JUC 的基石。