AQS 源码解读

文章列表

AQS 源码解读

一、AQS

AQS 是 AbstractQueuedSynchronizer 的简称，又称为同步阻塞队列，是 Java 中的一个抽象类。在其内部维护了一个由双向链表实现的 FIFO 线程等待队列，同时又提供和维护了一个共享资源 state ，像我们平常使用的 ReentrantLock、Semaphore、ReentrantReadWriteLock、SynchronousQueue、FutureTask等都是基于AQS 进行实现的。

AQS 可以实现什么功能呢？在 AQS 中不考虑资源的获取和释放，主要关注资源获取不到时，如何将线程加入队列以及阻塞，当释放资源后，如何再进行线程的出列和唤醒。而对于资源的操作则交予具体实现的子类进行完成。

在此基础上 AQS 为了使线程的控制可以更灵活，又提供了两种同步模型，独占模式和共享模式：

独占模式：表示并发情况下只有一个线程能执行，其余则需等待，例如 Lock 锁，一次只能有一个线程获取到锁。
共享模式：允许多线程根据规则执行，例如 Semaphore 进行多个线程的协调。

AQS 已经帮我们实现了队列的维护，以及线程的等待和唤醒，但是具体资源的获取和释放都需要由继承类实现，对于资源的获取和释放也是区分了独占模式和共享模式，相应方法如下：

//查询是否正在独占资源，condition会使用
boolean isHeldExclusively()	
//独占模式，尝试获取资源，成功则返回true，失败则返回false
boolean tryAcquire(int arg)
//独占模式，尝试释放资源，成功则返回true，失败则返回false
boolean tryRelease(int arg)
//共享模式，尝试获取资源，如果返回负数表示失败，否则表示成功。
int tryAcquireShared(int arg)
//共享模式，尝试释放资源，成功则返回true，失败则返回false。
boolean tryReleaseShared(int arg)

例如在 ReentrantLock 公平锁中，tryAcquire 的实现逻辑如下：

protected final boolean tryAcquire(int acquires) {// 当前线程final Thread current = Thread.currentThread();// AQS 中共享 stateint c = getState();if (c == 0) {// 如果队列中没有其他线程，并对state进行修改，// 如果修改成功则设置独占锁的线程为当前线程if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current == getExclusiveOwnerThread()) {// 如果独占线程就是当前线程，则是重入的场景，对 state + 1int nextc = c + acquires;if (nextc < 0)throw new Error("Maximum lock count exceeded");setState(nextc);return true;}// 如果都没成功，则获取锁失败return false;}
}

可以看到在 ReentrantLock 公平锁中，通过 state 的值来标识是否有锁资源可用，并且重入情况下也是对 state 的值进行修改标识，对于 state 的修改和判断是否有等待队列线程， AQS 中都提供了相应的方法。

在 AQS 中几个核心的方法如下，同样区分了独占模式和共享模式：

// 返回共享资源的当前值
final int getState()
// 设置共享资源的值
final void setState(int newState)
// CAS设置共享资源的值
final boolean compareAndSetState(int expect, int update)// 独占模式获取同步资源，会调用重写的tryAcquire(int arg)，
// 如果获取成功，则不做任何处理，否则将会加入同步队列并挂起线程等待
final void acquire(int arg)
// 独占模式式获取同步资源，但是可以响应中断
final void acquireInterruptibly(int arg)
// 独占模式获取同步资源，但多出了超时时间，
// 如果当前线程在 nanosTimeout 时间内没有获取到同步资源，
// 那么将会返回false，否则返回true
final boolean tryAcquireNanos(int arg, long nanosTimeout)
// 独占模式式释放同步资源，会调用重写的 tryRelease(int arg) 方法，
// 在释放同步资源之后，会将同步队列中第一个节点包含的线程唤醒
final boolean release(int arg)// 共享模式式获取同步资源，会调用重写的 tryAcquireShared(int arg) ，
// 如果当前线程未获取到同步资源，会加入同步队列等待，
// 和独占式的区别这里 tryAcquireShared(int arg) < 0 时才认为未获取到资源
final void acquireShared(int arg)
// 共享模式式获取同步资源，可以响应中断
final void acquireSharedInterruptibly(int arg)
// 共享模式获取同步资源，但多出了超时时间
final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
// 共享式释放同步资源，会调用重写的 tryReleaseShared(int arg) 方法，
// 在释放同步资源之后，会将同步队列中第一个节点包含的线程唤醒
final boolean releaseShared(int arg)

下面一起从源码的角度，分析 AQS 是如何实现线程的协调和管理的。

二、共享资源 state

共享资源 state 就是 AQS 中的一个 int 类型的全局变量，使用了 volatile 进行修饰，保证了多线程下的数据可见性，并且 AQS 为其提供了普通和 CAS 方式的修改方法，该共享资源主要用来做资源的标记。

例如：

在 ReentrantLock 锁中用来表示是否获取到锁，默认情况 0 表示无锁状态，获取到锁后进行 +1 ，如果是重入的场景下同样进行 +1 ，最后释放锁后再进行 -1。

在 Semaphore 中用来表示信号量的标记，当获取信号量时 state 进行 -1 ，释放信号量再进行 +1 ：

AQS 源码解读

三、FIFO 阻塞队列

在 AQS 中阻塞队列采用双向链表进行实现，具体源码如下：

	//等待队列节点类，双向链表static final class Node {// 标记，指示节点正在共享模式下等待static final Node SHARED = new Node();// 标记，指示节点正在独占模式下等待static final Node EXCLUSIVE = null;//  waitStatus值表示线程已取消static final int CANCELLED = 1;//  waitStatus值表示后继线程需要唤醒static final int SIGNAL = -1;//  waitStatus值，表示线程正在等待状态static final int CONDITION = -2;// waitStatus值指示下一个被获取的应该无条件的传播static final int PROPAGATE = -3;// 线程等待状态volatile int waitStatus;// 上一个节点volatile Node prev;// 下一个节点volatile Node next;// 当前线程volatile Thread thread;// 节点的模式，独占还是贡献Node nextWaiter;// 是否为共享模型final boolean isShared() {return nextWaiter == SHARED;}// 获取上一个节点final Node predecessor() throws NullPointerException {Node p = prev;if (p == null)throw new NullPointerException();elsereturn p;}Node() {    // Used to establish initial head or SHARED marker}Node(Thread thread, Node mode) {     // Used by addWaiterthis.nextWaiter = mode;this.thread = thread;}Node(Thread thread, int waitStatus) { // Used by Conditionthis.waitStatus = waitStatus;this.thread = thread;}}

其中通过 nextWaiter 表示当前线程是独占模型还是共享模式，线程的所属状态使用 waitStatus 来进行表示，其中包括：

默认值为 0，表示当前节点在 sync 队列中，等待着获取资源。
CANCELLED，值为1，表示当前的线程被取消。
SIGNAL，值为-1，释放资源后需唤醒后继节点。
CONDITION，值为-2，等待condition唤醒。
PROPAGATE，值为-3，工作于共享锁状态，需要向后传播，比如根据资源是否剩余，唤醒后继节点。

四、独占模式 acquire 获取资源

在 AQS acquire() 方法中，首先调用子类的 tryAcquire 获取资源，如果资源获取成功则不做任何处理，如果失败则首先使用 addWaiter 将当前线程加入到队列中，并指定 Node 的类型为独占模式：

AQS 源码解读

在 addWaiter 方法中，会将当前线程的 Node 加入到队列的尾端，如果尾节点为空或修改尾节点失败则进入到 enq 中使用自旋的方式修改：

AQS 源码解读

在 enq 方法中可以看出，当为节点为空时，也就是队列中无数据时，会初始化一个空的 Head 节点。

AQS 源码解读

再回到 acquire() 方法，加入队列后会进入到 acquireQueued 方法中，在该方法循环中如果当前节点的pred 上一个节点是 head 节点的话，那该节点不就是第一个节点吗，因为从上面就可以看出，初始情况下 head 是一个空的 node ，那 head 的下一个节点不就是第一个进入到队列的节点了，这种情况下遵循队列先进先出的原则，再次尝试是否能获取到资源，如果可以成功获取资源到则将当前节点置为 head 节点，同时再次将 head 节点置为空 node，此时线程也无需阻塞可以直接执行：

AQS 源码解读

但是如果当前节点的上一个节点不是 head 节点，或者没有获取到资源，则此时需要进行挂起阻塞，下面首先会触发 shouldParkAfterFailedAcquire 方法，这里先看后面的 parkAndCheckInterrupt 方法，该方法主要做了将线程挂起阻塞的作用，采用 LockSupport.park 进行线程的阻塞：

AQS 源码解读

再来看 shouldParkAfterFailedAcquire 方法就是控制当前线程是否需要挂起，这里就需要使用到 Node 中的 waitStatus，在该方法中有三种类型的判断：

如果当前是 SIGNAL 状态则可以直接挂起
当waitStatus大于 0 时，在 Node 中 waitStatus 大于 0 的状态就是 CANCELLED 状态，也就是标识线程被取消了，此时这种线程进行阻塞也就没有意义了，那就一直循环向上取线程未被取消的作为当前节点，继续执行。
当waitStatus小于等于 0 时，将状态置为 SIGNAL 类型

AQS 源码解读

后面当阻塞的线程被唤醒后，会继续在 acquireQueued 的循环中，不断找寻第一个入队的线程进行尝试获取资源操作。

五、独占模式 release 释放资源

在 release 方法中，首先会调用子类的 tryRelease 方法释放资源：

AQS 源码解读
然后会将当前的 head 节点传入 unparkSuccessor 方法中，在该方法中首先将该Node节点的 waitStatus 修改到默认的 0 值，然后获取到下一个节点，因为 head 节点始终保持为空节点，下一个节点才是真正的队列中第一个线程。但如果下一个节点为空的话，或者已经被取消了，则循环从 tail 节点向上找最前面正常的节点，最后直接使用 LockSupport.unpark 唤醒该节点的线程：