JAVA - AQS源码解读

本文主要是介绍JAVA - AQS源码解读，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

JAVA - AQS源码解读

前言

AQS全称AbstractQueuedSynchronizer。是Lock、CountDownLatch、Semaphore（信号量）的基础。这个AQS我在大约一年多以前看过一次，但是当时水平有限，根本没看懂什么意思，最后对于AQS的理解都是来自于网上的博文，可以说是吃人家嚼过的。。这次重读源码还是收获颇丰。

Node

看一下Node的结构：

//标识等待节点处于共享模式

static final Node SHARED = new Node();

//标识等待节点处于独占模式这里直接是个null啊，有点懵逼，不知道有什么讲究

static final Node EXCLUSIVE = null;

//由于超时或中断，节点已被取消

static final int CANCELLED = 1;

//表示下一个节点是通过park堵塞的，需要通过unpark唤醒

static final int SIGNAL = -1;

//表示线程在等待条件变量（先获取锁，加入到条件等待队列，然后释放锁，等待条件变量满足条件；只有重新获取锁之后才能返回）

static final int CONDITION = -2;

//表示后续结点会传播唤醒的操作，共享模式下起作用

static final int PROPAGATE = -3;

//等待状态：对于condition节点，初始化为CONDITION；其它情况，默认为0

volatile int waitStatus;

volatile Node prev;

volatile Node next;

//线程对象，这里我们可以看出来，一个Node其实就是一个线程的包装

volatile Thread thread;

//对于Condtion表示下一个等待条件变量的节点；其它情况下用于区分共享模式和独占模式；

Node nextWaiter;

这里对这个nextWaiter的概念还不是很了解，故去找了一些大神的博客看了一下：

独占模式：每次只能有一个线程能持有资源
共享模式：允许多个线程同时持有资源

这么说可能还是很抽象，借鉴大神的几个例子：

CountDownLatch的await方法可以在多个线程中调用，当CountDownLatch的计数器为0后，调用await的方法都会依次返回。也就是说多个线程可以同时等待await方法返回，因此它适合被设计成共享模式，因为它获取的是一个共享资源，资源在所有调用await方法的线程间共享；
ReentrantLock提供了lock和unlock方法，只允许一个线程获得锁，因此它适合被设计成独占模式，因为它获取的是一个独占资源，资源不能在调用lock方法的线程间共享；
Semaphore维护了一组许可，acquire方法获取许可，如果有可用的许可，方法返回，否则block可用看到，acquire获取到也是一个共享资源，只不过资源的数量有限制，因此它适合被设计成共享模式；
ReentrantReadWriteLock提供了读写锁，写操作是独占的，读操作是可以彼此共享的，因此它同时使用了独占和共享模式；

我们总结一下，说白了就是资源的可见性，资源所有线程可见，就是共享，否则独占。我们就拿出两个来看一下，ReentrantLock的设计就是独占，上面也说了，对于ReentrantLock来说，它要访问的资源就是前继节点（为什么我们下面会说到），这个只有后继节点访问就可以了。我们再看CountDownLatch，对它来说，资源就是计数器，多个线程都在等待计数器归零，所以这个就是共享的。这么说不知道对不对，如果有偏差希望指正。

acquire

这篇文章的分析，会结合一下子类的实现，我们以ReentrantLock为例。看一下Lock方法，非公平的情况下，先直接尝试获取锁，失败了就会调用acquire方法：

public final void acquire(int arg) {

if (!tryAcquire(arg) &&

acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

selfInterrupt();

}

很简单，先尝试获取，失败了返回，成功了把节点一个独占节点加入队列。我们看一下tryAcquire方法的子类实现（以后子类都默认ReentrantLock）最终会调用nonfairTryAcquire：

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()) {

//如果独占线程已经是当前线程，把state + acquires（其实就是加一）

int nextc = c + acquires;

if (nextc < 0) // overflow

throw new Error("Maximum lock count exceeded");

//更新状态

setState(nextc);

return true;

}

return false;

}

这个acquires就是1，代码比较简单，不多说了，简单提一下如果独占线程就是当前线程的情况，这个情况下，会把state加1，这里就是可重入的概念，我们在释放的时候再说。

addWaiter

我们看了tryAcquire和acquire，再看看如果尝试获取成功之后入队列的操作，先看addWaiter：

private Node addWaiter(Node mode) {

//根据传入的模式(独占or共享)创建Node对象，我们这里是独占

Node node = new Node(Thread.currentThread(), mode);

// Try the fast path of enq; backup to full enq on failure

Node pred = tail;

//这里如果不为空，说明前面还有线程在等待，尝试CAS入列，如果入列失败，则调用enq采用自选的方式入列

//这里也写了

if (pred != null) {

node.prev = pred;

//通过CAS更新tail节点

if (compareAndSetTail(pred, node)) {

pred.next = node;

return node;

}

enq(node);

return node;

}

我们注意下，上面在pred.next = node;这里，我们是通过CAS来更新tail，CAS成功才会进来，这是tail已经是新节点了，这时node.prev已经是旧的tail了，我们为了从头开始向后遍历，再把pred.next设置为新node。

这里我们看一下enq，注释也说了，这个是通过自旋的方式入队：

private Node enq(final Node node) {

for (;;) {

Node t = tail;

//由于采用lazy initialize,当队列为空时，需要进行初始化

if (t == null) { // Must initialize

//通过CAS设置head和tail节点

if (compareAndSetHead(new Node()))

tail = head;

} else {

//将node的前继节点设置为原tail节点

node.prev = t;

//CAS更新tail节点，更新成功则将原tail节点的后节点设置为node

if (compareAndSetTail(t, node)) {

//还是为了能从队列头开始遍历队列，设置一下next

t.next = node;

return t;

}

我们可以看到，AQS本质上改自自旋锁。

acquireQueued

到这addWaiter结束了，我们看一下acquireQueued：

final boolean acquireQueued(final Node node, int arg) {

boolean failed = true;

try {

boolean interrupted = false;

//仍然通过自旋，根据前面的逻辑，此处传入的为新入列的节点

for (;;) {

final Node p = node.predecessor();

//如果node的前一节点为head节点,而head节点为空节点，说明node是等待队列里排在最前面的节点

if (p == head && tryAcquire(arg)) {

//获取资源成功，将node设置为头节点,setHead清空节点属性thread,prev

setHead(node);

p.next = null; // help GC

failed = false;

//返回是否发生中断

return interrupted;

}

//如果acquire失败，是否要park，如果是调用parkAndCheckInterrupt

if (shouldParkAfterFailedAcquire(p, node) &&

parkAndCheckInterrupt())

interrupted = true;

}

} finally {

if (failed)

//只有循环中出现异常，才会进入该逻辑

cancelAcquire(node);

}

上面也有注释，如果需要park，parkAndCheckInterrupt最终调用LockSupport.park。看一下shouldParkAfterFailedAcquire：

//shouldParkAfterFailedAcquire会将前节点的状态改为Node.SIGNAL,

//接着在下一次循环中调用parkAndCheckInterrupt阻塞线程，注意，设置为SIGNAL之后并没有返回true而是false，这就是为什么是下一次循环中才会阻塞线程

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

int ws = pred.waitStatus;

if (ws == Node.SIGNAL)

* This node has already set status asking a release

* to signal it, so it can safely park.

return true;

if (ws > 0) {

* Predecessor was cancelled. Skip over predecessors and

* indicate retry.

//如果前一节点已取消，则往前找，直到找到一个状态正常的节点，其实就是从队列删除取消状态的节点

do {

node.prev = pred = pred.prev;

} while (pred.waitStatus > 0);

//更新next指针，去掉中间取消状态的节点

pred.next = node;

} else {

* waitStatus must be 0 or PROPAGATE. Indicate that we

* need a signal, but don't park yet. Caller will need to

* retry to make sure it cannot acquire before parking.

//更新pred节点的waitStatus为SIGNAL

compareAndSetWaitStatus(pred, ws, Node.SIGNAL);

}

return false;

}

我们看到线程自旋入队的过程中，如果失败了并且需要挂起就挂起，唤醒了再尝试，失败了再挂起。判断是否需要挂起的过程中，在特定情况下还会移除队列中取消状态的节点。

cancelAcquire

看完获取再看取消获取：

private void cancelAcquire(Node node) {

// Ignore if node doesn't exist

if (node == null)

return;

node.thread = null;

// Skip cancelled predecessors

//获取node的前向节点

Node pred = node.prev;

//如果发现前向节点状态为CANCELLED，则继续向前找，直到找到状态正常的节点，这里跟上面的park的时候一样

while (pred.waitStatus > 0)

node.prev = pred = pred.prev;

// predNext is the apparent node to unsplice. CASes below will

// fail if not, in which case, we lost race vs another cancel

// or signal, so no further action is necessary.

Node predNext = pred.next;

// Can use unconditional write instead of CAS here.

// After this atomic step, other Nodes can skip past us.

// Before, we are free of interference from other threads.

node.waitStatus = Node.CANCELLED;

//如果node为tail节点，则将pred更新为tail节点

if (node == tail && compareAndSetTail(node, pred)) {

//由于pred为新的尾节点，因此将其next设为null

compareAndSetNext(pred, predNext, null);

} else {

// If successor needs signal, try to set pred's next-link

// so it will get one. Otherwise wake it up to propagate.

int ws;

if (pred != head &&

((ws = pred.waitStatus) == Node.SIGNAL ||

(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&

pred.thread != null) {

Node next = node.next;

if (next != null && next.waitStatus <= 0)

compareAndSetNext(pred, predNext, next);

} else {

unparkSuccessor(node);

}

node.next = node; // help GC

}

先看一下如果要cancelAcquire的节点是尾节点，那么就把尾节点（入参node）的前继节点设置为尾节点。

再看一下如果不是尾节点的情况，这里比较复杂。我们看一下if里的条件：

pred不是head节点：如果pred为头节点，而node又被cancel，则node.next为等待队列中的第一个节点，需要unpark唤醒。
pred节点状态为SIGNAL或能更新为SIGNAL。
pred的thread变量不能为null。

我们这里来看一下SIGNAL的含义，看注释含义：

该节点的后继节点是阻塞的，所以当前节点在释放或取消时必须取消其后继，为了避免冲突，获取方法必须首先指示他们需要一个信号，然后重试原子获取，然后在失败时阻止。

再看上面的三个条件，我们就比较清楚了。如果是头结点需要取消，那么就要唤醒后面的节点。如果状态无法更新成SIGNAL，同样需要唤醒后继节点，因为根据SIGNAL节点的含义我们知道，SIGNAL节点后的节点是需要唤醒的，并且后继节点是阻塞的。如果不能更新成SIGNAL，那我们需要手动唤醒后面节点，当前节点同样不会影响后继节点去获取资源。

我们看一下unparkSuccessor(node)这里，这是唤醒node的后继节点：