标签：node Node AQS pred next 探究 线程 节点 底层
java.utils.concurrent 简称 JUC，是jdk跟并发有关的包，AQS在其下的 locks 包下，全称为 AbstractQueuedSynchronizer，是一个抽象类。之所以有名是因为它是很多并发类的基类，最常见的 ReentrantLock 就是基于 AQS。也是因为AQS的存在，我们可以很方便的写出一个自己的并发工具类，只需要自定义如何获取及释放 state 属性即可。
Node节点
这是 AQS 中的内部类，用来形成一条双向队列封装阻塞的线程，其内部的属性名一看便知其意图。
| prev | 上一个节点 |
| next | 下一个节点 |
| thread | 当前线程 |
AQS类维护着 head 和 tail，就是等待队列的头和尾，抓住头尾节点，这条链就算是确定了。除了 node，另外还有一个非常重要的一个属性：state，所以线程靠抢占这个state来取得锁。
acquire 获取锁
这个方法写的很简单，难的都在它调用的方法上。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}tryAcquire 自定义获取锁
可以发现这是一个没有代码的方法，不用想就知道等着子类来自定义实现，跟后面释放锁 tryRelease是一样的。
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}来看一下 ReentrantLock 是如何重写 tryAcquire 抢占锁的。
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState(); // 获取state
if (c == 0) { // 0 表示锁没有被任何线程占用
if (compareAndSetState(0, acquires)) { // 使用cas设置state
setExclusiveOwnerThread(current); // 设置成功，则设置锁的占用者为当前线程
return true;
}
}
else if (current == getExclusiveOwnerThread()) { // 如果state不为0，先不急，看看占锁的线程是不是当前同一个线程
int nextc = c + acquires; //如果是，则可以重入。state表示重入的次数
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}可以看到就是通过 cas 方式设置 state 属性来抢占锁；如果state不为0，则先判断当前线程和占用锁的线程是否为同一个：是则重入，否则执行后续addWaiter方法放入等待队列进行阻塞。
是否怀疑为什么只使用cas设置state，设置线程所有者 setExclusiveOwnerThread 为什么就可以直接设置无序考虑线程安全问题？个人认为：
- 虽然cas是乐观锁，但是底层调用的是cpu原子指令，尽量减少cas的使用提高性能
- 当cas设置完state后，其他线程只会被阻塞到等待队列，这不会造成线程安全问题
- 在AQS中有好几处这类的逻辑，看似线程不安全，其实逻辑上没有问题
addWaiter 添加到等待队列
当tryAcquire尝试设置state失败，也就是占用锁失败后，接下来就会把当前线程包装的node放入到等待队列中去。
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; // 将当前节点的prev设置为尾节点
if (compareAndSetTail(pred, node)) { // cas将tail指向当前节点
pred.next = node; // 将原来的尾结点的next设置为当前节点
return node;
}
}
enq(node); // 队列为空或设置尾节点失败走这里
return node;
}addWaiter要做的就是把封装了当前线程对象的node节点放在队列的尾部。这个设置需要改动 AQS 的tail属性，所以需要 cas 操作。下面再次看到了上面提到过的做法：没有“将原来的尾结点的next设置为当前node”设置成cas，后面会补上这个小坑。
enq方法
当等待队列为空时，会进入enq方法，会在头节点设置一个没有意义的空节点。想象一下，处在头节点的就是表示线程正在执行的，头节点本来就不需要排队，空节点就很合理。
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // 队列为空
if (compareAndSetHead(new Node())) //cas设置头节点为空节点
tail = head;
} else { // 队列不为空
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}除了队列为空， 其实 addWaiter 方法中cas设置尾结点 compareAndSetTail 方法失败也会进去enq方法。为什么设置尾结点会失败呢？说明当前线程没有抢到锁，正要将其node往队列的尾部塞的时候，cpu时间片到期，线程上下文切换来到了另外一个线程，它率先一步塞入尾部，那么这个线程就会cas失败。
所以enq才会逻辑判断列表是否会为空，不为空就是上面所说的情形，就会重复 addWaiter 方法中cas添加尾结点的操作。使用for循环就是考虑到上述情形一直发生。
acquireQueued 阻塞
添加完队列后，不用想就知道下一步就是阻塞线程啦。在这里会调用当前node的 predecessor 方法获取其前继节点。如果前继节点为是头节点，那么当前节点就是排队等待队列中的第一个，先让它调用 tryAcquire 再次尝试抢占锁。
- 如果能抢占成功就更好：那么头节点就是当前节点，之前的头节点的next置空，让其了无牵挂的走
- 如果抢占失败：调用 shouldParkAfterFailedAcquire 方法将前继节点的状态设置为正常，确保自己可以被唤醒
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);
}
}shouldParkAfterFailedAcquire 设置前置节点
这里就要说一下节点的等待状态 waitStatus 问题。
- 取消：1
- signal：-1 正常可以被唤醒
- condition：-2
- propagate：-3
- 新加入节点：0
所以说等待状态小于等于0才算是正常的。大于0就应该将其剔除出队列。
shouldParkAfterFailedAcquire就是在向上遍历，确保自己的前继节点的等待状态是有效的，不要因为前继节点的状态问题导致自己无法被唤醒。
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;
}parkAndCheckInterrupt 阻塞当前线程
做完装备工作，终于线程可以安心睡去了。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}可以看到阻塞非常简单，调用的就是 LockSupport的 park方法，返回的是线程是否被标记打断。如果其他线程想要打断当前线程，那么当前线程的该标记就是true，那么就是走到acquire方法里调用 selfInterrupt 方法，进行自我打断。
static void selfInterrupt() {
Thread.currentThread().interrupt();
}interrupted、isInterrupted、interrupt 区别
- interrupt：线程调用其他线程的Interrupt方法，那么其他线程就会标记为打断。仅仅标记而已
- isInterrupted：查看线程是否被打上了中断标记。不会清除中断标记
- Interrupted：同isInterrupted。区别是它会清除中断标记
为什么不使用 stop 方法打断
stop方法因为被jdk认定为不安全的方法，因为stop会立即中止线程的执行，导致线程的执行逻辑不完整。例如转账，小张刚转钱给小明，小张账户的钱少了，小明的账户上的钱还没加，就给stop了，要了命。合理的中断就应该是Interrupt，让线程自己去判断对于中断该如何去响应它。
再看 acquireQueued
这个方法没有想象中的简单。
如果当前节点的前继节点不是头节点，就调用 parkAndCheckInterrupt 方法阻塞其线程。所以所有的线程都不被阻塞在这里，等待释放锁被唤醒。唤醒后又进入这个死循环（就没有出去过），直到其前继节点是头节点并抢锁成功，才跳出这里的循环。
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);
}
}这里的 Interrupted 是可以取到 true 的，就是在线程将要阻塞的时候，其他线程想要中断它，那么这里返回的就是true，在acquire方法里就不用调用 selfInterrupt 了。
那么什么时候 failed 会取到 true 来执行 cancelAcquire 取消获取锁呢？
找来找去也就在阻塞的时候因为jvm的原因导致阻塞异常，也就是说这是代码健壮性考虑，能到这里的概率很小很小。
cancelAcquire 取消
- 首先会将node的thread设置为空。
- 其次检查node前驱是否是取消状态，是则循环跳过，一直为node找一个正常的前驱。
- 接着node.waitStatus 设置为CANCELLED。
- 判断node是否在尾部，是则tail指针前移到node前驱上，node前驱成为新的tail，其next指针（predNext）设置为null。
- 若node不是在尾部，判断其前驱是否是head以及是否是正常节点。node前驱不是head且正常节点，则将node后继链接到node前驱next指针（predNext）上（Node next = node.next;compareAndSetNext(pred, predNext, next);），从而使node被剔除。
- 若node不是在尾部，且node前驱是head，则唤醒node的后继。node前驱不是head，但是不正常节点（刚好被取消的），则也唤醒node的后继，这时的唤醒不是为了让node后继获取锁，而是为node的后继链接一个正常的前驱（node后继自旋判断阻塞时shouldParkAfterFailedAcquire，会链接一个正常的前驱）。
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
node.thread = null;
// Skip cancelled predecessors
Node pred = node.prev;
while (pred.waitStatus > 0)
//pred = pred.prev;
//node.prev = pred;
node.prev = pred = pred.prev;
//如果node的前驱也是取消节点，则pred.next就不是node
Node predNext = pred.next;
node.waitStatus = Node.CANCELLED;
// If we are the tail, remove ourselves.
//如果node在尾部，tail前移
if (node == tail && compareAndSetTail(node, pred)) {
//node设置为null
compareAndSetNext(pred, predNext, null);
} else {
//node不在尾部
// 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)
//node后继成为pred的后继
compareAndSetNext(pred, predNext, next);
} else {
// 如果node的前驱是个head，则唤醒node后继，
//node前继节点不是一个正常的节点，唤醒后继节点
unparkSuccessor(node);
}
node.next = node; // help GC
}
}unparkSuccessor 唤醒线程
这里跟阻塞一样，release方法调用tryRelease，由子类自定义释放 state，其他交给AQS。
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}unparkSuccessor里的node参数就是头节点。
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.
*/
//唤醒后继节点的线程，若为空，从tail往后遍历找一个距离head最近的正常的节点
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);
}这里有个比较奇怪的逻辑，当头节点的下一个节点是为null或者状态大于0，也就是说这个节点无效时，就从等待列表尾部开始遍历，找到状态正常 <=0 的节点，唤醒它里面的线程。
之所以逆向遍历，是因为要补上面说的坑：在 addWaiter 中cas设置尾节点中，当前节点的prev已经设置为原来的尾结点了，cas设置tail的指向为当前节点成功后，t.next = node; 还没来得及设置，开始上下文开始切换并释放，这就造成了尴尬的局面：这个原先的尾结点没有next。如果从头遍历的尾，就会出现无法遍历完所有链表的情况。而从尾到头就没有这个问题。
同样的，在enq方法里，也有一个片段使用了 addWaiter 的设置尾结点的方法，所以跟enq方法也有关。
标签：node,Node,AQS,pred,next,探究,线程,节点,底层 来源： https://www.cnblogs.com/mebi/p/16350374.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。