-
JDK成长记19：ReenranctLock（2）加锁入队的AQS底层原理
上一节，你应该学到了ReentrantLock底层基于AQS的3个小组件 state、owner、queue。并且了解了下一个线程1进行加锁修改owner和state的过程。还记得么？加锁成功后，如下图所示的状态：
首次加锁的时候，只使用到了owner和state这两个小组件，并没有涉及到等待队列。所以这一节，我们继续看一下，如果有下一个线程—线程2，这个哥们过来加锁会是如何的？
直接从JDK源码层面理解AQS的另一个线程也来加锁的入队逻辑
直接从JDK源码层面理解AQS的另一个线程也来加锁的入队逻辑
当线程2这个哥们进行加锁的时候，假设线程1还没有释放锁，也就是基于上面的图的状态，线程2进行加锁。同样会走到如下lock方法的代码：
static final class NonfairSync extends Sync {
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
如果线程2进行lock，当执行compareAndSetState(0,1)的时候，由于state此时已经是1了，肯定会CAS操作失败，计入else逻辑，在NonFairSync的父类AQS中可以找到如下代码：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
// 接着又会调用NonFairSync实现的tryAcquire：
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
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()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
这个上面的tryAcquire方法实际调用了一个nonfairTryAcquire，从名字上看，叫做非公平获取的一个方法。（后面讲非公平锁的会讲到）。
但是当你看过这个方法的脉络你会发现，state是1，第一个if不满足，owner是线程1，当前是线程2，第二个if也不满足，结果直接返回了false。
所以到这里你会发现线程2加锁，截止到现在，会执行到如下图所示步骤3所示：
接着由于tryAcquire返回false，会进入&&后面的方法调用addWaiter(Node.EXCLUSIVE)。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
addWaiter从名字上，你可以连蒙带猜下，其实这个方法的意思就是添加到等待队列的进行等待的意思。让我们来看下：
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;
}
首先传入的是一个Node mode，就是Node.EXCLUSIVE,从名字上看就是一个独占Node的意思。
你可以这个Node.EXCLUSIVE看下：
static final class Node {
/** Marker to indicate a node is waiting in shared mode */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;
/** 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;
}
果然，你可以看到Node中有一堆静态变量，通过null，空Node、1、1、-2、-3表示一些Node的角色类型。
接着往下看addWaitder方法：
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
//省略后面代码
return node;
}
这个new Node又做了什么？可以看下Node的构造方法和成员变量：
volatile Node prev;
volatile Node next;
volatile int waitStatus;
volatile Thread thread;
Node nextWaiter;
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
除了next、prev、thread表示双向链表的前后指针和对应的数据元素之外，还有两个变量nextWaiter和waitStatus。可以从名字上猜出来，表示等待节点和等待状态的意思。
这里传入了thread=线程2，mode= EXCLUSIVE = null 。其实nextWaiter这里更像是个标记，表示独占类型的Node。或者说是线程2正在等待的是一个独占锁。创建的node如下图所示：
接着addwaiter创建完成节点node后，继续执行代码pred指针指向tail，但是默认tail是null，所以直接调用enq(node)方法，看样子是要进行入队。enqueue的意思。 代码如下：
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;
}
执行到这里就会得到如下结果：
AQS的本质：为啥叫做异步队列同步器？
AQS的本质：为啥叫做异步队列同步器？
接着我们需要分析下enq(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;
}
}
}
}
从脉络上看，是一个经典for循环+CAS 自旋操作。你可以跟着看下代码执行的思路：
1）第一次for循环
首先t指向tail，tail由于是null，t刚开始肯定是null，进入第一个if。
接着通过CAS操作compareAndSetHead，将head指向了新建的一个Node，成功后将tail指向了head。
private final boolean compareAndSetHead(Node update) {
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}
所以会得到如下图所示结果：
2）第二次for循环
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;
}
}
}
}
此时ReentrantLock的tail和head已经指向了空的new Node()。
接着还是t=tail, t此时不为空了。走到了else逻辑，使用入参node节点的prev指向了t所指向的空Node。
之后通过CAS操作compareAndSetTail,将tail指向到入参node节点。
private final boolean compareAndSetTail(Node expect, Node update) {
return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
最后通过t的next也指向了入参node节点。
也就是如下图所示：
从上图，我们就可以看出来，线程2的node和空node连接起来，形成了一个双向链表。之前学习LinkedList你应该已经知道，双向链表也可以当做队列使用。所以这里你可以当做将node进行了入队操作。
这个其实就是AQS的本质，等待队列组件的作用。
当线程2进行了入队等待，这里你可以简化一下流程图，你可以得到如下的图：
加锁失败的时候如何借助AQS异步入队阻塞等待？
加锁失败的时候如何借助AQS异步入队阻塞等待？
入队后，接着就结束了么？不是，还需要修改下线程2的状态，将他进行挂起，既然已经排上队了，就不要占用CPU资源了，是不是？
我们看下是如何做的：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
之前我们执行完了addWaiter,返回的节点是node，也就是线程2对应的等待节点，arg是1。接着进入了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);
}
}
目前等待队列情况如下：
这个方法核心脉络，是一个无限for循环，当中有两个if。
接着我们看下细节：
1）第一次For循环：
首先上来使用一个辅助指针p，指向了node节点的前一个节点，node.predecessor其实就是p=node.prev。代码如下：
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
由于head等于p,就还是尝试获取一次锁，tryAcquire(arg)。这里假设线程1还没有释放锁，tryAcquire(arg)肯定还是会失败返回false，所以第一个if不成立。（如果获取成功，这个if其实会将线程2移出队列的）
接着执行第二个if判断，先进行了shouldParkAfterFailedAcquire方法调用，第一个参数传入p，就是空Node,第二参数传入node，就是线程2对应的node。
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);
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.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
第一个参数传入p，就是空Node,第二参数传入node，就是线程2对应的node。
它们两个节点的waitStatus都是0。所以经过上面代码，会执行到最后一个else。
会通过CAS操作，将空Node的waitStatus状态(ws)从0改为Node.SIGNAL(-1)。如下图所示：
接着shouldParkAfterFailedAcquire就直接返回false，第一个条件false。就会直接进行下一次for循环了。
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);
}
}
2）第二次For循环
假设线程1还是没有释放锁，上面的for循环还是会进入如下方法，但是其实的pred也就是空Node的watiStatus已经被改成SIGNAL（-1）,所以之里会返回true。
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);
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.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
接着下面这个if第一个条件是ture会判断第二条件parkAndCheckInterrupt
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
parkAndCheckInterrupt这个方法从名字看叫做挂起并且检查线程是否被打断。代码如下
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
可以看到他核心调用了一个工具类LockSupport.park(this);
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, 0L);
setBlocker(t, null);
}
这个底层是通过UNSAFE的C++代码实现的，我们就不去看了。你只要知道，这个park操作会将线程挂起，进入等待状态就可以了。还记得之前将线程的状态图么？
park操作会将线程挂起，进入Waiting等待状态。也就是说线程2加锁失败最终就是入队并且等待。
今天这一节，到这里就把AQS中入队的逻辑给大家讲清楚了。线程获取锁失败如何入队？如何挂起的？相信你都很清楚了。你可以自己用第三个线程尝试加锁失败彻底图解AQS队列等待机制试试。最后学完，如果你可以画出这个图，就说嘛你真正明白了AQS的基本原理了。
小结&思考
小结&思考
虽然这个入队逻辑看着比较复杂，但其实大家可以抽象出这个队列的设计是基于：CAS操作+Node状态+线程标记控制就可以了。
可以多思考下关键思想和关键点，不用太纠结细节。比如多思考下为啥设计了状态，是为了单独使用Condition吗？还是。。。。
这些思考才是最重要的！
下一节，我们看下如果线程1释放了锁，如何唤醒队列中元素的。唤醒的时候如果有本地线程来加锁，还能插队！？所以下一节也会给大家介绍下什么是公平和非公平锁。
来源：https://www.cnblogs.com/fanmao/p/15482058.html
