Java并发-AQS核心原理深度剖析

AQS 核心原理深度剖析

AQS 是 JUC 包的基石，理解 AQS 就理解了 Java 并发工具类的一半。

一、什么是 AQS？

AQS（AbstractQueuedSynchronizer，抽象队列同步器）是 Doug Lea 设计的一个用于构建锁和同步器的框架。它抽象了锁的核心要素：状态管理 + 线程排队 + 阻塞唤醒。

1.1 基于 AQS 的实现类

同步器使用模式说明
ReentrantLock独占可重入互斥锁
ReentrantReadWriteLock独占 + 共享读写分离锁
Semaphore共享信号量，控制并发数
CountDownLatch共享倒计数门闩
CyclicBarrier独占循环屏障

1.2 核心设计思想

AQS 采用模板方法模式，将通用逻辑（入队、出队、阻塞、唤醒）封装在父类，将同步状态的获取与释放逻辑留给子类实现。

1
2
3
4
5
6
// 子类需要实现的方法
protected boolean tryAcquire(int arg)      // 独占获取
protected boolean tryRelease(int arg)      // 独占释放
protected int tryAcquireShared(int arg)    // 共享获取
protected boolean tryReleaseShared(int arg) // 共享释放
protected boolean isHeldExclusively()      // 是否被当前线程独占

二、AQS 核心组件

2.1 同步状态（state）

1
2
3
4
5
6
7
8
9
// volatile 保证可见性
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);
}

state 在不同锁中的含义：

锁类型state 含义
ReentrantLock锁的重入次数（0 表示未锁定）
Semaphore可用许可数
CountDownLatch剩余计数值
ReadWriteLock高 16 位：读锁持有数；低 16 位：写锁重入数

2.2 CLH 同步队列

AQS 维护了一个变体 CLH（Craig, Landin, and Hagersten）双向链表作为同步队列：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
static final class Node {
    // 节点状态
    volatile int waitStatus;
    
    // 双向链表指针
    volatile Node prev;
    volatile Node next;
    
    // 节点持有的线程
    volatile Thread thread;
    
    // 条件队列的下一个节点 / 共享模式标识
    Node nextWaiter;
    
    // 共享模式标识
    static final Node SHARED = new Node();
    // 独占模式标识
    static final Node EXCLUSIVE = null;
    
    // waitStatus 取值
    static final int CANCELLED =  1;  // 取消
    static final int SIGNAL    = -1;  // 后继节点需要被唤醒
    static final int CONDITION = -2;  // 在条件队列中等待
    static final int PROPAGATE = -3;  // 共享模式下需要传播唤醒
}

队列结构示意图：

1
2
3
4
5
6
     +------+  prev  +------+       +------+
head |      | <----- |      | <---- |      | tail
     | Node |  next  | Node |       | Node |
     |      | -----> |      | ----> |      |
     +------+        +------+       +------+
      哨兵节点         等待节点        等待节点

注意：head 节点是一个哨兵节点（dummy node），不存储线程信息。

2.3 等待状态详解

waitStatus值说明
SIGNAL-1当前节点释放锁时需要唤醒后继节点
CANCELLED1节点被取消（超时或中断）
CONDITION-2节点在条件队列中等待
PROPAGATE-3共享模式下，释放需要传播
初始值0新节点的初始状态

三、独占模式源码分析

3.1 获取锁流程

1
2
3
4
5
6
7
8
public final void acquire(int arg) {
    // 1. tryAcquire 尝试获取锁（子类实现）
    // 2. 失败则 addWaiter 加入同步队列
    // 3. acquireQueued 自旋 + 阻塞
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();  // 恢复中断标志
}

addWaiter：入队

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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);
    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;
            }
        }
    }
}

acquireQueued：自旋 + 阻塞

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 只有前驱是 head 才有资格尝试获取锁
            if (p == head && tryAcquire(arg)) {
                setHead(node);  // 获取成功，成为新的 head
                p.next = null;  // help GC
                failed = false;
                return interrupted;
            }
            // 判断是否需要阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())  // LockSupport.park() 阻塞
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);  // 取消获取
    }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;  // 前驱已设置 SIGNAL，可以安心阻塞
    if (ws > 0) {
        // 前驱已取消，向前遍历找到有效节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 将前驱状态设置为 SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

3.2 释放锁流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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)
        compareAndSetWaitStatus(node, ws, 0);
    
    // 找到需要唤醒的后继节点
    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);  // 唤醒线程
}

四、共享模式

4.1 获取共享锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)  // 返回值 < 0 表示获取失败
        doAcquireShared(arg);
}

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 设置头节点并传播唤醒
                    setHeadAndPropagate(node, r);
                    p.next = null;
                    if (interrupted) selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed) cancelAcquire(node);
    }
}

4.2 独占模式 vs 共享模式

对比维度独占模式共享模式
典型实现ReentrantLockSemaphore、CountDownLatch
获取结果booleanint（剩余资源数）
唤醒方式仅唤醒一个后继链式唤醒多个后继
适用场景互斥访问并发访问控制

五、Condition 条件队列

5.1 工作原理

每个 Condition 对象维护一个单向链表（条件队列），与同步队列配合实现等待/通知机制。

1
2
3
4
5
6
7
同步队列（双向链表）                条件队列（单向链表）
+------+     +------+             +------+     +------+
| head | <-> | Node | ...         | first| --> | Node | ...
+------+     +------+             +------+     +------+
                 |                    ^
                 | await()            | signal()
                 +--------------------+

5.2 await() 流程

  1. 将当前线程封装成 Node 加入条件队列
  2. 完全释放锁（state 清零）
  3. 阻塞等待被 signal() 唤醒
  4. 被唤醒后重新竞争锁

5.3 signal() 流程

  1. 将条件队列的首节点移动到同步队列尾部
  2. 唤醒该节点的线程重新竞争锁

六、AQS 设计精髓

  1. 模板方法模式：框架负责排队、阻塞、唤醒，子类只需实现 try 方法
  2. CLH 队列变体：自旋 + 阻塞的混合策略，平衡 CPU 和响应
  3. volatile + CAS：无锁化实现状态管理和队列操作
  4. 双向链表：便于取消节点的处理
  5. SIGNAL 状态：避免无效唤醒，提高效率

七、要点

Q1：AQS 的核心是什么？

state 同步状态 + CLH 同步队列 + 模板方法模式

Q2：为什么用双向链表？

方便处理取消的节点，从尾部向前遍历更安全（入队时先设置 prev）

Q3：独占模式和共享模式的区别？

独占只允许一个线程持有；共享允许多个线程同时持有，且会链式传播唤醒