详解AbstractQueuedSynchronizer（AQS）源码

引言

上篇文章讲解了CountDownLatch源码，底层是继承了AQS基类调用父类和重写父类方法实现的，本文将简介AQS源码和架构设计，帮助我们更深入理解多线程实战。

源码架构

1.  状态变量 state
AQS 使用一个 int 类型的变量 state 来表示同步状态。这个变量是 volatile 修饰的，保证了多线程之间的可见性。不同的同步器对 state 有不同的定义和使用方式，例如在 ReentrantLock 中，state 表示锁的重入次数；在 CountDownLatch 中，state 表示计数器的值。

private volatile int state;

2. 双向队列
AQS 内部维护了一个双向链表队列，用于存储等待获取锁的线程。这个队列是 CLH（Craig, Landin, and Hagersten）锁队列的变种。每个节点（Node）代表一个等待线程，包含线程引用、等待状态等信息。

static final class Node {
    // 共享模式
    static final Node SHARED = new Node();
    // 独占模式
    static final Node EXCLUSIVE = null;

    // 线程已取消
    static final int CANCELLED =  1;
    // 后继节点需要被唤醒
    static final int SIGNAL    = -1;
    // 线程在等待条件
    static final int CONDITION = -2;
    // 表示下一次共享式同步状态获取将会无条件地传播下去
    static final int PROPAGATE = -3;

    // 等待状态
    volatile int waitStatus;
    // 前驱节点
    volatile Node prev;
    // 后继节点
    volatile Node next;
    // 节点所包装的线程
    volatile Thread thread;
    // 指向下一个处于 CONDITION 状态的节点
    Node nextWaiter;
    ...
}

而双向队列细分为条件队列和同步队列

同步队列

同步队列是 AQS 的核心队列，用于存放等待获取锁的线程节点。当线程尝试获取锁失败时，会被封装成节点加入到这个队列中等待。

在 AQS 类中有两个关键属性用于表示同步队列的头节点和尾节点：

private transient volatile Node head;
private transient volatile Node tail;

head 指向队列的头节点，tail 指向队列的尾节点，这两个属性都是 volatile 修饰的，保证了多线程环境下的可见性。

节点加入同步队列（addWaiter 方法）

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 方法插入
    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;
            }
        }
    }
}

条件队列

条件队列用于实现线程的等待 - 通知机制，一个 AQS 可以有多个条件队列，每个 Condition 对象对应一个条件队列。

关键类 ConditionObject

ConditionObject 是 AQS 中实现条件队列的内部类，它包含一个指向条件队列头节点和尾节点的引用

public class ConditionObject implements Condition, java.io.Serializable {
    private transient Node firstWaiter;
    private transient Node lastWaiter;
    ...
}

线程进入条件队列（await 方法）

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // 如果尾节点已取消，清除队列中所有已取消的节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

唤醒条件队列中的线程

public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

final boolean transferForSignal(Node node) {
    // 将节点的等待状态从 CONDITION 改为 0
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    // 将节点从条件队列转移到同步队列
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}