标签：node JUC return pred acquire Node 源码 发包 节点
JUC并发包源码阅读之ReetrantLock.lock()
我们从代码最顶层开始看起，ReetrantLock默认实现的是NonfairSync，NonfairSync的lock（）方法如下，非常简单
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
if (compareAndSetState(0, 1)) //尝试cas获取锁,如果state为0,获取成功
setExclusiveOwnerThread(Thread.currentThread());
else //state不为0
acquire(1);
}
lock()方法就是先尝试cas的将state从0设为1,但是如果state不为0,或者cas自旋失败,调用acquire()方法,这里需要注意,在FairSync中lock（）方法的实现是直接调用acquire（）方法，并不会先尝试cas自旋获取锁，这也体现了公平和非公平的差异。
接下来看看acquire（）方法
public final void acquire(int arg) {
if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
我们先根据这个方法中调用的方法的名字来猜测一下这个acquire()方法是实现了什么功能,tryAcquire（）就是尝试获取的意思，addWaiter()就是插入一个节点到队尾的意思，acquireQueued（）就是获取队列的意思，在揣测一下就是唤醒整个队列。由此我们可以大概猜出这个acquire的方法就是当尝试获取锁失败并且线程的中断标志位为true，就中断线程。
tryAcquire（）实际调用的是nonfairTryAcquire（）方法，接下来详细看看nonfairTryAcquire()方法
/**
* Performs non-fair tryLock. tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) { //state为0成功
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()方法中，是需要同步队列中没有节点或者节点位于队列头并且cas更新state成功才会获得锁。
当nonfairTryAcquire（）方法失败时，也就是没有获取到锁时，会调用addWaiter（）方法将节点插入同步队列，接下来看addWaiter()方法。
/**
* Creates and enqueues node for current thread and given mode.
*
* @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
* @return the new 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) { //aqs队尾不为空,cas将node插入队列尾部
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node); //aqs队尾为空或cas尾插失败
return node;
}
当同步队列的队列为空(因为需要cas更新队尾,如果队列为空无法进行cas更新队尾)或者尾插失败(可能由于多个节点同时插入队尾)就会调用enq()方法自旋尾插。
/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize //aqs队列为空就新建一个
if (compareAndSetHead(new Node()))
tail = head;
} else { //尾插失败就不断cas重试
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
节点插入队尾addWaiter（）后会调用acquireQueued（）方法
/**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
*/
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)) { //节点在AQS队列中的第二个（head节点的下一个）并释放锁成功，将其设为head
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) && //如果前一个节点的waitStatus为singal，park当前线程，否则将前一个节点的waitStatus设为singal
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
这是一个不断循环的方法,直到节点为head节点的下一个节点并释放锁成功时,将其设为头节点,也就是说节点获取到锁了,才会退出循环。否则循环调用shouldParkAfterFailedAcquire()方法。
/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/
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; //前一个节点的waitStatus为signal时才返回true，否则将前一个节点的waitStatus设为signal然后返回false
}
这个方法也是十分简单，我们需要记住这个方法的放回值，只有当前一个节点的waitStatus为SIGNAL才放回true，否则返回false。
由上面的acquireQueued（）方法可知，当shouldParkAfterFailedAcquire（）方法返回true时，会调用 parkAndCheckInterrupt()方法。
/**
* Convenience method to park and then check if interrupted
*
* @return {@code true} if interrupted
*/
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
这个方法的作用就是park住当前线程并返回当前线程是否被中断。
当acquireQueued（）方法失败时，会执行cancelAcquire()方法，将节点移除队列。
/**
* Cancels an ongoing attempt to acquire.
*
* @param node the node
*/
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) //前一个节点的ws大于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;
// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) { //node是尾节点,把前一个节点设为尾节点
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) { //如果前一个节点不是head,并且前一个节点为signal(或者cas设置signal成功)
Node next = node.next;
if (next != null && next.waitStatus <= 0) //后一个节点不为空并且ws小于0,删除该节点
compareAndSetNext(pred, predNext, next);
} else {
unparkSuccessor(node); //前一个节点是head或者前一个节点的ws大于0,释放该线程
}
node.next = node; // help GC
}
}
思维导图
仅供参考
标签：node,JUC,return,pred,acquire,Node,源码,发包,节点 来源： https://blog.csdn.net/weixin_45003125/article/details/111312599
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。