AQS 详解
AQS原理分析
在 synchronized 详解中已经提到管程和 MESA 模型，管程中引入了条件变量的概念，而且每个条件变量都对应有一个等待队列。条件变量和等待队列的作用是解决线程之间的同步问题。
Java 中针对管程有两种实现方式：
- 一种是基于 Object 的 Monitor 机制，用于
synchronized内置锁的实现。 - 另一种是抽象队列同步器 AQS，用于 JUC 包下 Lock 锁机制的实现。不同的是
synchronized只有一个等待队列，而 Lock 可以有多个等待队列。
什么是AQS
java.util.concurrent 包中的大多数同步器实现都是围绕着共同的基础行为，比如同步队列、等待队列、独占获取、共享获取等，而这些行为的抽象就是基于 AbstractQueuedSynchronizer（简称 AQS）实现的。AQS 是一个抽象同步框架，可以用来实现一个依赖状态的同步器。
JDK 中提供的大多数同步器（如 Lock, Latch, Barrier 等），都是基于 AQS 框架来实现的。
- 一般是通过一个内部类
Sync继承 AQS。 - 将同步器所有调用都映射到
Sync对应的方法中。
AQS具备的特性：
- 阻塞同步队列
- 独占/共享
- 公平/非公平
- 可重入
- 允许中断
AQS核心结构
//共享变量，使用volatile修饰保证线程可见性
private volatile int state;
// 返回同步状态的当前值
protected final int getState() {
return state;
}
// 设置同步状态的值
protected final void setState(int newState) {
state = newState;
}
// 原子地（CAS操作）将同步状态值设置为给定值update，如果当前同步状态的值等于expect（期望值）
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}AQS 内部维护了属性 volatile int state：
state表示资源的可用状态。state的三种访问方式：getState()setState()compareAndSetState()
定义了两种资源访问方式：
- Exclusive（独占）：只有一个线程能执行，如
ReentrantLock。 - Shared（共享）：多个线程可以同时执行，如
Semaphore/CountDownLatch。
AQS 在自定义实现时主要需实现以下几种方法：
isHeldExclusively()：该线程是否正在独占资源。只有用到 condition 才需要去实现它。tryAcquire(int)：独占方式。尝试获取资源，成功则返回 true，失败则返回 false。tryRelease(int)：独占方式。尝试释放资源，成功则返回 true，失败则返回 false。tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0 表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回 true，否则返回 false。
AQS定义两种队列
- 同步队列：主要用于维护获取锁失败时入队的线程。
- 等待队列：调用
await()的时候会释放锁，然后线程会加入到等待队列；调用signal()唤醒的时候会把等待队列中的线程节点移动到同步队列中，等待再次获得锁。
同步队列
AQS 当中的同步队列也称 CLH 队列。CLH 队列是 Craig、Landin、Hagersten 三人发明的一种基于双向链表数据结构的队列，是 FIFO 先进先出的线程同步队列。Java 中的 CLH 队列是原 CLH 队列的一个变种，线程由原自旋机制改为了阻塞机制。
AQS 依赖 CLH 同步队列来完成同步状态的管理：
- 当前线程如果获取同步状态失败时，AQS 则会将当前线程及等待状态等信息构造成一个节点（Node），并将其加入到 CLH 同步队列中，同时阻塞当前线程。
- 当同步状态释放时，会把首节点唤醒（公平锁），使其再次尝试获取同步状态。
- 通过
signal或signalAll将等待队列中的节点转移到 同步队列。（由等待队列转化为同步队列）
等待队列
AQS 中等待队列是使用单向链表保存的，用 nextWaiter 来连接：
- 调用
await方法阻塞线程； - 当前线程存在于同步队列的头结点时，调用
await方法进行阻塞，会从同步队列转化到 等待队列。
AQS 内部通过 waitStatus 变量来记录节点（Node）的状态，主要包含以下 5 种：
- 0：普通等待状态。表示节点在同步队列中正常等待获取锁，没有其他特殊标记。
- CANCELLED (1)：取消状态。节点因等待超时或中断而被取消，此后状态不再变化，并会被移出同步队列。
- SIGNAL (-1)：唤醒接力状态。表示当前节点的后继节点处于等待状态（或即将阻塞），当当前节点释放了同步状态或被取消时，需要负责唤醒（unpark）其直接后继节点。
- CONDITION (-2)：条件等待状态。表示节点当前处于 等待队列 中，未在 同步队列 中。当调用 Condition 的
signal()方法后，该节点会从等待队列转移到同步队列中，重置状态为 0 并重新竞争锁。 - PROPAGATE (-3)：传播状态。仅用于共享锁场景（
ReentrantLock等独占锁不使用），表示释放锁时唤醒操作需要向后传播，避免唤醒链条断裂，确保后续共享节点能依次被唤醒。
基于AQS实现一把独占锁
思考：基于 AQS 如何设计一把简单的独占锁（支持可重入）？
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
/**
* 自定义可重入独占锁，基于 AQS 实现。
* 支持同一个线程多次获取锁，需释放相同次数才能完全释放锁。
*/
public class JuziLock extends AbstractQueuedSynchronizer {
/**
* 尝试获取锁（独占模式）。
* 实现可重入逻辑：
* - 若锁未被占用（state == 0），则通过 CAS 将 state 置为 1，并设置当前线程为锁持有者。
* - 若锁已被当前线程持有，则 state 加 1，表示重入次数增加。
* - 其他情况返回 false，表示获取失败。
*
* @param unused 未使用的参数，仅用于符合 AQS 方法签名
* @return 成功获取锁返回 true，否则返回 false
*/
@Override
protected boolean tryAcquire(int unused) {
Thread current = Thread.currentThread();
// 获取当前同步状态（锁重入次数）
int c = getState();
if (c == 0) {
// 锁未被任何线程占用，尝试 CAS 获取锁
if (compareAndSetState(0, 1)) {
// 记录锁持有线程
setExclusiveOwnerThread(current);
return true;
}
} else if (current == getExclusiveOwnerThread()) {
// 锁已被当前线程持有，增加重入计数
int nextc = c + 1;
// 防止重入次数溢出（理论上不会发生）
if (nextc < 0) {
throw new Error("Maximum lock count exceeded");
}
setState(nextc);
return true;
}
// 锁被其他线程持有，获取失败
return false;
}
/**
* 尝试释放锁（独占模式）。
* 实现可重入释放：
* - 将重入计数减 1。
* - 只有当计数减到 0 时，才真正释放锁（清空锁持有者）。
* - 否则仅更新 state，表示仍持有锁。
*
* @param unused 未使用的参数，仅用于符合 AQS 方法签名
* @return 如果锁已完全释放（state == 0）则返回 true，否则返回 false
*/
@Override
protected boolean tryRelease(int unused) {
// 校验当前线程是否为锁持有者，不是则抛出异常
if (Thread.currentThread() != getExclusiveOwnerThread()) {
throw new IllegalMonitorStateException();
}
// 重入次数减 1
int c = getState() - 1;
boolean free = false;
if (c == 0) {
// 重入次数归零，表示锁完全释放
free = true;
// 清空锁持有线程
setExclusiveOwnerThread(null);
}
// 更新同步状态
setState(c);
return free;
}
/**
* 获取锁（阻塞式）。如果锁已被其他线程持有，当前线程将进入等待队列。
*/
public void lock() {
acquire(1);
}
/**
* 尝试获取锁（非阻塞）。立即返回结果，不会进入等待队列。
*
* @return 成功获取锁返回 true，否则返回 false
*/
public boolean tryLock() {
return tryAcquire(1);
}
/**
* 释放锁。如果当前线程持有锁多次，则仅减少一次重入计数；
* 当重入计数减至 0 时，才真正唤醒等待队列中的下一个线程。
*/
public void unlock() {
release(1);
}
/**
* 判断锁是否已被任何线程占用（包括当前线程）。
*
* @return 锁被占用返回 true，否则返回 false
*/
public boolean isLocked() {
return getState() != 0;
}
}使用测试：
package com.juzicoding.aqs;
public class JuziLockExample {
private static int counter = 0;
private static final JuziLock lock = new JuziLock();
public static void main(String[] args) throws InterruptedException {
testReentrancy();
testConcurrent();
testTryLock();
}
private static void testReentrancy() {
lock.lock();
try {
System.out.println("第一次获取锁");
lock.lock(); // 重入
try {
System.out.println("第二次获取锁（重入）");
} finally {
lock.unlock();
System.out.println("释放一次锁");
}
} finally {
lock.unlock();
System.out.println("释放所有锁");
}
}
private static void testConcurrent() throws InterruptedException {
Thread[] threads = new Thread[10];
for (int i = 0; i < threads.length; i++) {
threads[i] = new Thread(() -> {
for (int j = 0; j < 1000; j++) {
lock.lock();
try {
counter++;
} finally {
lock.unlock();
}
}
});
threads[i].start();
}
for (Thread t : threads) {
t.join();
}
System.out.println("计数器最终值（应为 10000）: " + counter);
}
private static void testTryLock() throws InterruptedException {
Thread holder = new Thread(() -> {
lock.lock();
try {
System.out.println("子线程持有锁");
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock();
System.out.println("子线程释放锁");
}
});
holder.start();
Thread.sleep(100); // 确保子线程先拿到锁
if (lock.tryLock()) {
try {
System.out.println("主线程 tryLock 成功");
} finally {
lock.unlock();
}
} else {
System.out.println("主线程 tryLock 失败，锁已被占用");
}
holder.join();
}
}输出结果：
第一次获取锁
第二次获取锁（重入）
释放一次锁
释放所有锁
计数器最终值（应为 10000）: 10000
子线程持有锁
主线程 tryLock 失败，锁已被占用
子线程释放锁ReentrantLock源码分析
ReentrantLock 是一种基于 AQS 框架的应用实现，是 JDK 中的一种线程并发访问的同步手段。它的功能类似于 synchronized 是一种互斥锁，可以保证线程安全。
ReentrantLock原理
ReentrantLock 是基于 AQS + CAS 实现的。
lock()流程图
ReentrantLock 基于抽象队列同步器 AQS + CAS 实现了加锁、释放锁的过程。ReentrantLock 实现了公平锁与非公平锁。公平锁与非公平锁唯一的区别在于：
非公平锁不会判断同步队列中是否有节点正在等待获取锁，而是直接尝试获取锁。这种行为被称为 Barging（插队）机制，它为非公平锁提供了两次插队获取锁的机会：
- 第一次：在
lock()方法开始执行时，直接尝试通过 CAS 修改state。 - 第二次：如果第一次失败，进入
acquire(1)时会调用tryAcquire，此时仍会再次尝试插队获取锁。
只有当这两次“插队”都失败后，非公平锁才会乖乖进入同步队列，一旦入队，其行为也将遵循 FIFO（先进先出）原则。相较于公平锁，这种设计虽然可能导致某些线程“饥饿”，但通过减少线程挂起的概率，能显著提升系统的整体吞吐量。
unlock()流程图
ReentrantLock 释放锁的流程较为简单：优先判断持有锁资源的线程是否为当前线程，若不为当前线程则抛出异常；若为当前线程，则将 AQS 的 state 的属性值减 1，再判断减 1 后的值是否为 0。若为 0，表示当前线程彻底释放了锁资源，随后会唤醒同步队列中的挂起线程节点，开始抢占锁资源。
源码分析核心方法
构造函数
private final Sync sync;
// 默认使用非公平锁
public ReentrantLock() {
sync = new NonfairSync();
}
// fair=true，公平锁；否则，非公平锁
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}Sync 是 ReentrantLock 的抽象静态内部类，继承自 AQS (AbstractQueuedSynchronizer) —— 抽象队列同步器。AQS 中定义了锁的基本行为，其中用 volatile 修饰的 state 表示当前锁重入的次数。
NonfairSync、FairSync 是 ReentrantLock 的静态内部类，继承了 ReentrantLock.Sync。NonfairSync 实现非公平锁，FairSync 实现公平锁。
lock()加锁
private final Sync sync;
// 加锁
public void lock() {
sync.lock();
}公平锁
调用 AQS 的 acquire 方法。ReentrantLock.FairSync#lock() 核心代码：
// 加锁
final void lock() {
acquire(1);
}非公平锁
通过 CAS 尝试获取锁（将 AQS 的 state 由 0 修改为 1）。若成功，代表当前线程获取锁资源成功；若失败，则调用 AQS 的 acquire 方法。ReentrantLock.NonfairSync#lock() 核心代码：
// 加锁
final void lock() {
// 获取锁资源，CAS 修改 AQS 的 state 属性值，获取成功，设置当前线程
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
// 获取失败，执行AQS的acquire
else
acquire(1);
}acquire()
acquire() 方法是 Sync 父类 AQS 中的方法，AbstractQueuedSynchronizer#acquire() 核心代码：
// 获取锁资源
public final void acquire(int arg) {
// 尝试获取锁资源
if (!tryAcquire(arg) &&
// 当前线程未获取到锁资源，加入同步队列，同时挂起线程，等待唤醒
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}tryAcquire()
tryAcquire() 方法在 FairSync、NonfairSync 中均有实现，用于尝试获取锁资源。核心代码如下：
// 公平锁 FairSync#tryAcquire() 方法
protected final boolean tryAcquire(int acquires) {
// 获取当前线程
final Thread current = Thread.currentThread();
// 获取AQS的 state
int c = getState();
// state == 0 当前没有线程占用锁资源
if (c == 0) {
// 判断是否有线程在排队，若无线程在排队，则尝试抢锁
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
// 抢锁成功，将线程属性设置为当前线程
setExclusiveOwnerThread(current);
return true;
}
}
// state != 0 有线程占用锁资源
// 判断占用锁资源的线程是否为当前线程
else if (current == getExclusiveOwnerThread()) {
// state + 1 (锁重入)
int nextc = c + acquires;
// 锁重入超出最大限制 (int的最大值)，抛出异常
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
// 将 state + 1 设置给 state
setState(nextc);
// 当前线程拿到锁资源，返回true
return true;
}
return false;
}
// 非公平锁 NonfairSync#tryAcquire() 方法
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
// 非公平锁 Sync#nonfairTryAcquire() 方法
final boolean nonfairTryAcquire(int acquires) {
// 获取当前线程
final Thread current = Thread.currentThread();
// 获取AQS的 state
int c = getState();
// 无线程占用锁资源
if (c == 0) {
// CAS 修改 state 的值，修改成功，设置线程属性为当前线程，返回占用锁资源标识
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
// 有线程占用锁资源
// 判断占用锁资源的线程是否是当前线程（重入）
else if (current == getExclusiveOwnerThread()) {
// AQS 的 state + acquires
int nextc = c + acquires;
// 超出锁重入的上限(int的最大值)，抛出异常
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
// 将 state + acquires 设置到 state 属性
setState(nextc);
return true;
}
return false;
}获取当前线程及 AQS 的 state：
- 若 AQS 的
state属性值为 0，表示无线程占用锁资源，公平锁会判断同步队列中是否有线程在排队：若有线程在排队，则返回尝试抢锁失败标识，随后进入入队逻辑。若无排队则 CAS 抢锁。 - 若
state属性值不为 0，判断持有锁资源的线程是否为当前线程。若为当前线程，AQS 的state属性值累加，返回尝试抢锁成功标识。
公平锁与非公平锁的整体实现流程类似，唯一不同的是：当 AQS 的 state 属性值为 0（无线程占用锁资源）时，非公平锁不会判断是否有线程在同步队列中排队，而是直接通过 CAS 抢占锁。
addWaiter()
为当前线程创建入队节点 AbstractQueuedSynchronizer.Node，入参 mode 表示锁类型。在 AQS 的静态内部类 Node 中有 SHARED、EXCLUSIVE 两个常量，SHARED 代表共享锁、EXCLUSIVE 代表排它锁。
AbstractQueuedSynchronizer#addWaiter() 核心代码：
// 同步队列的尾节点，懒加载，只能通过enq方法添加节点
private transient volatile Node tail;
private Node addWaiter(Node mode) {
// 将当前线程、获取的锁类型封装为Node对象
Node node = new Node(Thread.currentThread(), mode);
// 获取同步队列的尾节点
Node pred = tail;
// 尾节点不为null
if (pred != null) {
// 将当前节点的前驱设置为同步队列的尾节点
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
// 同步队列为空或CAS抢占队尾失败，进入enq初始化并自旋入队
enq(node);
// 返回当前线程节点
return node;
}同步队列不为空时，尝试将当前线程封装的 Node 节点添加进队列尾部；若同步队列为空，则先初始化同步队列，然后再将 Node 节点添加进队列尾部。
enq()
同步队列尾节点为空时，执行 enq() 方法初始化同步队列，并将 Node 节点添加进同步队列中。
private Node enq(final Node node) {
for (;;) {
// 获取同步队列的尾节点
Node t = tail;
// 同步队列为空，初始化同步队列
if (t == null) {
// 初始化同步队列头尾节点
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 当前线程的Node添加到同步队列尾部中
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}acquireQueued()
判断当前线程是否需要挂起，AbstractQueuedSynchronizer#acquireQueued() 核心代码：
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)) {
// 将当前节点设置为 head - 头节点
setHead(node);
// 原头节点的 next 指向设置为null，便于GC回收
p.next = null;
// 设置获取锁资源成功
failed = false;
// 返回中断状态
return interrupted;
}
// 如果当前节点不是head的下一个节点，或者获取锁资源失败，尝试将线程挂起
if (shouldParkAfterFailedAcquire(p, node) &&
// 线程挂起，调用 UNSAFE.park()
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}查看当前排队的 Node 是否是 head 的 next 节点。如果是，尝试获取锁资源；如果不是，或者获取锁资源失败，那么就尝试将当前 Node 对应的线程挂起（调用 unsafe.park()）。
shouldParkAfterFailedAcquire 检查并更新未成功获取锁资源的节点状态，返回 true 表示线程需要被挂起。
AbstractQueuedSynchronizer#shouldParkAfterFailedAcquire() 核心代码：
static final class Node {
// 线程被取消
static final int CANCELLED = 1;
// 表示当前节点的后继节点需要被唤醒
static final int SIGNAL = -1;
// 表示当前节点处于等待队列中
static final int CONDITION = -2;
// 传播状态，用于共享锁场景
static final int PROPAGATE = -3;
}
// 获取锁资源失败，判断是否挂起线程
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 获取当前节点的前驱节点的状态
int ws = pred.waitStatus;
// 如果前驱节点状态为 SIGNAL，表示前驱释放锁时会通知当前节点，可以放心挂起
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
// 前驱节点被取消，向前查找并跳过所有已取消的节点
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 前驱节点状态为 0 或 PROPAGATE，通过 CAS 设置为 SIGNAL，
// 标记当前节点需要被唤醒，暂不挂起，外层会自旋重试
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}在挂起线程前，需确认当前节点的上一个节点的状态。若为 1，代表是已取消的节点，不能挂起；若为 -1，代表后续节点中有需要挂起的线程，可以放心挂起；若为 -2 (线程在 等待队列中) 或 -3 (避免共享锁线程无法唤醒的一个状态)，需要将状态改为 -1 之后，才能挂起当前线程。
unlock()释放锁
释放锁，ReentrantLock#unlock() 核心代码：
// 释放锁
public void unlock() {
sync.release(1);
}unlock 方法实际调用的是 AQS 的 release 方法，AbstractQueuedSynchronizer#release() 核心代码：
// 同步队列的头节点，懒加载，通过setHead方法初始化
private transient volatile Node head;
// 释放锁
public final boolean release(int arg) {
// 尝试释放锁资源
if (tryRelease(arg)) {
// 当前线程完全释放锁资源后，获取同步队列头节点
Node h = head;
if (h != null && h.waitStatus != 0)
// 唤醒同步队列中待唤醒的下一个节点
unparkSuccessor(h);
// 完全释放锁资源
return true;
}
// 当前线程未完全释放锁资源（可能存在重入）
return false;
}tryRelease() 尝试释放锁，ReentrantLock.Sync#tryRelease() 的核心代码：
// 释放锁
protected final boolean tryRelease(int releases) {
// 修改 AQS 的 state 计数值
int c = getState() - releases;
// 当前线程不是持有锁的线程，抛出异常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
// 是否成功的将锁资源完全释放标识 （state == 0）
boolean free = false;
// 锁资源完全释放
if (c == 0) {
// 修改标识
free = true;
// 将占用锁资源的属性设置为null
setExclusiveOwnerThread(null);
}
// 重新更新 state 赋值
setState(c);
// 返回true表示当前线程完全释放锁资源；
// 返回false表示当前线程仍持有锁资源，只是持有计数值减少
return free;
}