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
}
}
思维导图
仅供参考