深入理解 AQS - AbstractQueuedSynchronizer

1. AQS
- 1.1 什么是 AQS
- 1.2 AQS 具备的特性
2. AQS 原理解析
- 2.1 AQS 原理概述
- - 2.1.1 什么是 CLH 锁
  - 2.1.2 AQS 中的队列
- 2.2 AQS 共享资源的方式：独占式和共享式
- - 2.2.1 Exclusive（独占式）
  - 2.2.2 Share（共享式）
- 2.3 AQS 底层使用了模板方法模式
- 2.4 AQS 定义了两种队列
- - 2.4.1 AQS 定义了5个队列中节点状态
  - 2.4.2 同步等待队列
  - 2.4.3 条件等待队列
  - - Condition 接口
3. AQS 源码分析
- 3.1 ReentrantLock 概述
- 3.2 创建重入锁
- 3.3 公平锁/⾮公平锁的 lock()
- - 3.3.1 公平锁 NonfairSync.lock() 方法
  - 3.3.2 非公平锁 NonfairSync.lock() 方法
  - 3.3.3 acquire()
  - 3.3.4 FairSync.tryAcquire(arg)
  - 3.3.5 NonFairSync.tryAcquire(arg)
  - 3.3.6 addWaiter(Node.EXCLUSIVE)
  - 3.3.7 acquireQueued(...)
- 3.4 解锁
- 3.4.1 tryRelease()
- 3.4.2 unparkSuccessor

1. AQS

在介绍 AQS 之前，先一起回顾一下 CAS（Compare And Swap）。

1.1 什么是 AQS

AQS，全名 AbstractQueuedSynchronizer，是一个抽象同步队列，它的内部通过维护一个状态 volatile int state（共享资源的状态），一个 FIFO 线程等待队列来实现同步功能。

Java 并发包很多工具类底层都是基于 AQS 来实现的，比如：Lock 工具 ReentrantLock、栅栏 CountDownLatch、信号量 Semaphore 等。

1.2 AQS 具备的特性

阻塞等待队列
共享/独占
公平/非公平
可重入
允许中断

2. AQS 原理解析

2.1 AQS 原理概述

AQS 内部维护 state 属性，state 用关键字 volatile 修饰，代表着该共享资源的状态一更改就能被所有线程可见，而 AQS 的加锁方式本质上就是多个线程在竞争 state，当 state 为 0 时代表线程可以竞争锁，不为 0 时代表当前对象锁已经被占有，其他线程来加锁时则会失败，加锁失败的线程会被放入一个 FIFO 的等待队列中，这些线程会被 UNSAFE.park() 操作挂起，等待其他获取锁的线程释放锁才能够被唤醒。

2.1.1 什么是 CLH 锁

由 Craig、Landin 和 Hagersten 三位大佬发明，因此命名为 CLH 锁；
是单向链表实现的队列；
是一个自旋公平（FIFO）锁，能确保无饥饿性；
申请线程只在本地变量上自旋，它不断轮询前驱的状态，如果发现前驱节点释放了锁就结束自旋。

一起看一下线程加锁的过程：

首先获得当前线程的当前节点 curNode，这里每次获取的 CLHNode 节点的 locked 状态都为false；
然后将当前 CLHNode 节点的 locked 状态赋值为 true，表示当前线程的一种有效状态，即获取到了锁或正在等待锁的状态；

注意，为了保证 locked 属性线程间可见，该属性被 volatile 修饰。

线程对 tail 域调用 getAndSet 方法，使自己成为队列的尾部，同时获取一个指向其前趋结点的引用 PreNode；
因为尾指针 tailNode 的总是指向了前一个线程的 CLHNode 节点，因此这里利用尾指针 tailNode 取出前一个线程的 CLHNode 节点，然后赋值给当前线程的前继节点 preNode，并且将尾指针重新指向最后一个节点即当前线程的当前 CLHNode 节点，以便下一个线程到来时使用；
根据前继节点（前一个线程）的 locked 状态判断，若 locked 为 false，则说明前一个线程释放了锁，当前线程即可获得锁，不用自旋等待；若前继节点的 locked 状态为 true，则表示前一线程获取到了锁或者正在等待，自旋等待。

在这里插入图片描述

如上图：有3个并发线程同时启动执行 lock 操作，假如3个线程的实际执行顺序为：T1、T2、T3

线程1 过来，执行了 lock 操作，获得了锁，此时 locked 状态为 true
线程2 过来，执行了 lock 操作，由于线程1 还未释放锁，此时自旋等待，locked 状态也为 true
线程3 过来，执行了 lock 操作，由于线程2 处于自旋等待，此时线程3 也自旋等待（因此 CLH 锁是公平锁），locked 状态也为 true

以下为 CLH 锁的释放锁过程：

首先从当前线程的线程本地变量中获取出当前 CLHNode 节点，同时这个 CLHNode 节点被后面一个线程的 preNode 变量指向着；
然后将 locked 状态置为 false 即释放了锁；

注意：locked 因为被 volitile 关键字修饰，此时后面自旋等待的线程的局部变量 preNode.locked 也为 false，因此后面自旋等待的线程结束 while 循环即结束自旋等待，此时也获取到了锁。这一步骤也在异步进行着。

然后给当前线程的表示当前节点的线程本地变量重新赋值为一个新的 CLHNode。

2.1.2 AQS 中的队列

AQS 中的队列是 CLH 变体的虚拟双向队列，通过将每条请求共享资源的线程封装成一个节点来实现锁的分配：

在这里插入图片描述
AQS 中的 CLH 变体等待队列拥有以下特性：

AQS 中队列是个双向链表，也是 FIFO 先进先出的特性
通过 head、tail 头尾两个节点来组成队列结构，通过 volatile 修饰保证可见性
head 指向节点为已获得锁的节点，是一个虚拟节点，节点本身不持有具体线程
获取不到同步状态，会将节点进行自旋获取锁，自旋一定次数失败后会将线程阻塞，相对于 CLH 队列性能较好

AQS 使用一个 int 成员变量来表示同步状态，通过内置的 FIFO 队列来完成获取资源线程的排队工作。AQS 使用 CAS 对该同步状态进行原子操作实现对其值的修改：

// 共享变量，使用 volatile 关键字修饰保证线程可见性
private volatile int state;

资源的可用状态通过 protected 类型的 getState、setState、compareAndSetState 进行操作：

// 返回同步状态的当前值
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);
}

2.2 AQS 共享资源的方式：独占式和共享式

AQS 定义了两种资源共享方式：独占式（Exclusive）和共享式（Share）。

2.2.1 Exclusive（独占式）

只有一个线程能执行，如 ReentrantLock。又可分为公平锁和非公平锁，ReentrantLock 同时支持两种锁。

2.2.2 Share（共享式）

多个线程可以同时执行，如 Semaphore/CountDownLatch。

2.3 AQS 底层使用了模板方法模式

同步器的设计是基于模板方法模式的，如果需要自定义同步器一般的方式是这样（模板方法模式很经典的一个应用） :

使用者继承 AbstractQueuedSynchronizer 并重写指定的方法（这些重写方法很简单，无非是对于共享资源 state 的获取和释放）。
将 AQS 组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。

这和我们以往通过实现接口的方式有很大区别，这是模板方法模式很经典的一个运用，感兴趣的可以参考：https://refactoringguru.cn/design-patterns/template-method

AQS 使用了模板方法模式，自定义同步器时需要重写下面几个 AQS 提供的模板方法：

isHeldExclusively()：该线程是否正在独占资源。只有用到 condition 才需要去实现它。
tryAcquire(int)：独占方式。尝试获取资源，成功则返回 true，失败则返回 false。
tryRelease(int)：独占方式。尝试释放资源，成功则返回 true，失败则返回 false。
tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0 表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回 true，否则返回 false。

2.4 AQS 定义了两种队列

同步等待队列：主要用于维护获取锁失败时入队的线程。
条件等待队列：调用 await() 的时候会释放锁，然后线程会加入到条件队列，调用 signal()/signalAll() 唤醒的时候会把条件队列中的线程节点移动到同步队列中，等待再次获得锁。

2.4.1 AQS 定义了5个队列中节点状态

值为0，初始化状态，表示当前节点在 sync 队列中，等待着获取锁。
CANCELLED，值为 1，表示当前的线程被取消；
SIGNAL，值为 -1，表示当前节点的后继节点包含的线程需要唤醒，也就是 unpark；
CONDITION，值为 -2，表示当前节点在等待 condition，也就是在 condition 队列中；
PROPAGATE，值为 -3，表示当前场景下后续的 acquireShared 能够得以执行；


/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED =  1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL    = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
 * waitStatus value to indicate the next acquireShared should
 * unconditionally propagate
 */
static final int PROPAGATE = -3;

2.4.2 同步等待队列

AQS 当中的同步等待队列是 CLH 的变体，见：2.1.1、2.1.2，

当前线程如果获取同步状态失败时，AQS 则会将当前线程已经等待状态等信息构造成一个节点（Node）并将其加入到 CLH 同步队列，同时再一次尝试获取锁，如果获取失败则会阻塞当前线程；
当同步状态释放时，会把首节点唤醒（公平锁），使其再次尝试获取同步状态；
通过 signal/signalAll 将条件队列中的节点转移到同步队列。（由条件队列转化为同步队列）。

在这里插入图片描述

2.4.3 条件等待队列

AQS 中条件队列是使用单向列表保存的，用 nextWaiter 来连接：

调用 await 方法阻塞线程；
当前线程存在于同步队列的头结点，调用 await 方法进行阻塞（从同步队列转化到条件队列）

Condition 接口

在这里插入图片描述

调用 Condition#await 方法会释放当前持有的锁，然后阻塞当前线程，同时向 Condition 队列尾部添加一个节点，所以调用 Condition#await 方法的时候必须持有锁。
调用 Condition#signal 方法会将 Condition 队列的首节点移动到阻塞队列尾部，然后唤醒因调用 Condition#await 方法而阻塞的线程（唤醒之后这个线程就可以去竞争锁了），所以调用 Condition#signal 方法的时候必须持有锁，持有锁的线程唤醒被因调用 Condition#await 方法而阻塞的线程。

3. AQS 源码分析

3.1 ReentrantLock 概述

ReentrantLock 是可重入的独占锁，只能有一个线程可以获取该锁，其它获取该锁的线程会被阻塞而被放入该锁的阻塞队列里面。当我们想要使⽤重⼊锁的时候，使⽤⽅式⼀般是如下3个步骤：

public class ReentrantLockTest {

    /** 1. 创建重入锁  */
    ReentrantLock lock = new ReentrantLock();

    public void doSomething() {
        // 2. 加锁 block until condition holds
        lock.lock();
        
        try {
            // do something
        } catch (Exception e) {
            // ...
        } finally {
            // 3. 解锁
            lock.unlock();
        }
    }
}

后面我们就是针对这3个步骤对其源码进⾏解析。在此之前，先一起看一下 Sync、FairSync、NonfairSync 是在哪⾥被使⽤的。

ReentrantLock 默认采用非公平锁，因为考虑获得更好的性能，通过 boolean 来决定是否用公平锁（传入 true 用公平锁）。

/** Synchronizer providing all implementation mechanics */
private final Sync sync;

/**
 * Creates an instance of {@code ReentrantLock}.
 * This is equivalent to using {@code ReentrantLock(false)}.
 */
public ReentrantLock() {
    sync = new NonfairSync();
}

/**
 * Creates an instance of {@code ReentrantLock} with the
 * given fairness policy.
 *
 * @param fair {@code true} if this lock should use a fair ordering policy
 */
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

观察源码可以发现 FairSync、NonfairSync 都是继承 Sync，而 Sync 又继承于 AbstractQueuedSynchronizer。

static final class FairSync extends Sync

static final class NonfairSync extends Sync

abstract static class Sync extends AbstractQueuedSynchronizer

3.2 创建重入锁

ReentrantLock 默认是非公平的：

public ReentrantLock() {
    sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

当我们使⽤⽆参构造⽅法去创建重⼊锁的时候，底层使⽤的是⾮公平锁

当⼊参 fair 等于 false 的时候，采⽤的就是⾮公平锁 - NonfairSync
当⼊参 fair 等于 true 的时候，采⽤的就是公平锁 - FairSync

3.3 公平锁/⾮公平锁的 lock()

3.3.1 公平锁 NonfairSync.lock() 方法

公平锁直接就执行 acquire() 排队等待：

/**
 * 公平锁
 */
static final class FairSync extends ReentrantLock.Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
        // 执行等待
        acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // 和非公平锁相比，这里多了一个是否有线程在等待的判断
            if (!hasQueuedPredecessors() &&
                    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;
    }
}

3.3.2 非公平锁 NonfairSync.lock() 方法

/**
 * 非公平锁
 */
static final class NonfairSync extends ReentrantLock.Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * 立即执行抢占锁操作。如果抢占成功，则加锁；如果抢占失败，则等待
     */
    final void lock() {
        // AQS的state如果成功被设置为1，则表示本线程已抢占锁成功；否则，抢占锁失败
        if (compareAndSetState(0, 1))
            // 将 AOS.excLusive0wnerThread 设置为当前线程
            setExclusiveOwnerThread(Thread.currentThread());
        else
            // 执行等待
            acquire(1);
    }

    // 是否获得非公平锁
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

先判断是否有锁，0 是无锁、1 是有锁，等于 0 的话则修改成 1 并且获得锁：

// 如果当前状态值等于期望值 0，则自动将同步状态设置为给定的更新值 1
protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

设置当前独占锁的拥有者线程：

protected final void setExclusiveOwnerThread(Thread thread) {
    // 将当前线程设置到这把独占锁上
    exclusiveOwnerThread = thread;
}

抢锁失败就进行则等待：

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

公平锁和非公平锁只有两处不同：

非公平锁在调用 lock 后，首先就会调用 CAS 进行一次抢锁，如果这个时候恰巧锁没有被占用，那么直接就获取到锁返回了。
非公平锁在 CAS 失败后，和公平锁一样都会进入到 tryAcquire 方法，在 tyAcquire 方法中，如果发现锁这个时候被释放了（state == 0），非公平锁会直接 CAS 抢锁，但是公平锁会判断等待队列是否有线程处于等待状态，如果有则不去抢锁，乖乖排到后面。

非公平锁如果这两次 CAS 都不成功，那么后面和公平锁是一样的，都要进入到阻塞队列等待唤醒。

相对来说，非公平锁会有更好的性能，因为它的吞吐量比较大。当然，非公平锁让获取锁的时间变得更加不确定，可能会导致在阻塞队列中的线程长期处于饥饿状态。

3.3.3 acquire()

⽆论是公平锁还是⾮公平锁，他们都有机会调⽤相同的⽅法，即：acquire(1)

/**
 * 以独占模式获取，忽略中断 (interrupts)
 * 通过调用至少一次 tryAcquire(int) 来实现，成功返回。否则线程排队，可能重复阻塞和解除阻塞，调用 tryAcquire(int) 直到成功
 * 此方法可用于实现方法 Lock.lock()
 */
public final void acquire(int arg) {
    /**
     * tryAcquire(1): 进行抢锁操作，返回是否抢锁成功
     * addWaiter(Node.EXCLUSIVE): 构建一个独占式节点 Node，维护好该节点的前后指针 Node
     * acquireQueued(addWaiter(Node.EXCLUSIVE), 1) 判断获取锁失败的时候，是否应该挂起该线程，如果是，则挂起当前线
     */
    if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // 设置当前线程的中断标识
        selfInterrupt();
}

也就是分为三步 tryAcquire(尝试抢锁)、addWaiter(构建节点加入到队列中)、acquireQueued(获取锁失败的时候将线程挂起)

3.3.4 FairSync.tryAcquire(arg)

/**
 * 进行抢锁操作，返回是否成功
 *
 * case1> 没人抢占锁（state==0），线程A尝试执行抢占锁操作，如果抢占成功，则返回true; 如果抢占失败，则返回false。
 * case2> 有人已经抢占了这个锁（state != 0），但是抢占这个锁的线程就是自己，那么对自己执行重入加锁操作，返回true；如果不是自己抢占的锁，那么返回false。
 */
protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    /** case1: 如果c等于0，说明可以抢占锁 */
    if (c == 0) {
        // 如果线程不需要排队 && 抢占锁成功（即:如果state=0，则将该值修改为1，CAS操作成功）
        if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    /** case2: 如果c不等于0，判断是否是重入操作 (即: 锁本来就是被自己抢占的，支持多次抢占) */
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

通过获取 state 来判断锁是否被获取，等于 0 的话不要排队直接获取锁，可以看到比 NonFairSync 多了 hasQueuedPredecessors 是否需要进入队列排队。

如果不等于 0，判断是否是重入；否则返回 false。

/**
 * 主要是用来判断线程需不需要排队。true：线程需要排队。false：线程不需要排队。
 * 因为队列是 FIF0 的，所以需要判断队列中有没有相关线程的节点已经在排队了。有则返回true表示线程需要排队；没有则返回 false 表示线程无需排队；
 */
public final boolean hasQueuedPredecessors() {
    // 读取头节点
    Node t = tail;
    // 读取尾节点
    Node h = head;
    // s是首节点h的后继节点
    Node s;

    /**
     * h != t -> 队列中有 >= 2个节点
     * (s = h.next) == null -> 头节点没有后继节点，即：只有自己一个node或者创建了h的后置节点，但是还没有执行h.next=node
     * s.thread != Thread.currentThread() -> 第二个节点承载的线程不是当前线程
     */
    return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());
}

3.3.5 NonFairSync.tryAcquire(arg)

/**
 * 是否获得非公平锁
 */
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}

/**
 * 进行抢锁操作，是否抢到非公平锁
 *
 * 处理内容：
 * 1>如果抢到锁，返回true
 *   1.1>如果当前线程第一次抢到锁:
 *        AQS.state由0变为1
 *        AQS.exclusiveOwnerThread=Thread.currentThread()
 *        返回true
 *   1.2>如果当前线程再次抢到锁(重入加锁):
 *        AQS.status++
 *        返回true
 * 2>如果没抢到锁，返回false
 */
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        /**
         * 获得当前独享线程，如果就是当前线程，那么执行重入操作
         * 执行tryLock()时：
         *      如果第二次进入，则 nextc = 0 + 1 = 1
         *      如果第三次进入，则 nextc = 1 + 1 = 2
         *      如果第四次进入，则 nextc = 2 + 1 = 3
         */
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

3.3.6 addWaiter(Node.EXCLUSIVE)

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

3.3.7 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)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

3.4 解锁

public void unlock() {
    sync.release(1);
}

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

3.4.1 tryRelease()

首先执行tryRelease，再执行 unpark 操作

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

3.4.2 unparkSuccessor

private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    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);
}