上文简单描述了ReentrantLock中的方法大部分都是由AbstractQueuedSynchronizer或它的子类实现，AQS队列和条件队列节点持有的信息，和它们运行时的数据结构是什么样子的。
本文结合最常用的lock和unlock方法来描述AQS队列是怎么样构建和运作的。先看一个简单的demo.
public class LockDemo extends Thread {
private Lock lock = new ReentrantLock();
public LockDemo(Lock lock){
this.lock = lock;
}
public void run() {
try {
this.lock.lock();
System.out.println("我获取到锁了："+Thread.currentThread().getName());
} finally {
this.lock.unlock();
}
}
public static void main(String[] args) {
Lock lock = new ReentrantLock();
LockDemo lockDemo1 = new LockDemo(lock);
lockDemo1.setName("lock1");
LockDemo lockDemo2 = new LockDemo(lock);
lockDemo2.setName("lock2");
LockDemo lockDemo3 = new LockDemo(lock);
lockDemo3.setName("lock3");
lockDemo1.start();
lockDemo2.start();
lockDemo3.start();
}
}
输出的结果如下，如我们所料，控制台是串行输出了结果：
我获取到锁了：lock1
我获取到锁了：lock2
我获取到锁了：lock3
先简单画下lock()方法代码流程图。
lock.png
ok,下面开始聊实现细节。ReentrantLock默认是以非公平模式构建，后面会简单说下公平模式和非公平模式构建的区别。废话 不多说，开始NonfairSync.lock()。
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
先用CAS尝试设置锁的状态，成功则把持有锁的线程设置为当前线程。否则执行acquire(1)，这里为什么是1，因为是ReentrantLock是可重入锁，同一个线程每次获取锁都会+1。接着看aqs.accquire;
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire是由AQS的子类，这里也就是NonfairSync去实现tryAcquire。
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;
}
第一步判断当前锁是否被持有，如果没有被持有则cas设置锁的状态，更新持有锁的线程。如果锁已经被持有，则进行第二步判断，如果持有锁的线程是当前线程，则进行重入锁的计算。上面两步都没有成功，则该线程持有锁的动作失败。开始构建AQS队列。看下acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法，先看addWaiter(Node.EXCLUSIVE)。
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队列尾节点，是不是为空，如果不为空则直接把当前节点CAS操作为尾节点。否则执行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循环，退出去的条件就是当前节点置位尾节点成功。先判断当前AQS队列是否存在尾节点。不存在说明队列不存在。则CAS操作生成一个头节点，并且置位尾节点。注意该节点并不持有任何线程信息，就是一个空节点。如果尾节点不为空。则把当前节点置位尾节点，之前的尾节点的next置位当前节点。从而构造了一个双向链表。构造节点成功后，开始执行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循环，退出循环的条件为把当前节点置位头结点（头结点不包含持有线程），首先获取当前节点的前置节点，如果当前节点是AQS队列第一次构建出来的，那么它的前置节点显然就是头节点。此时会去尝试持有锁，如果持有锁成功了，把当前 节点置位头结点。并把之前的头结点的next置为null，减少强引用方便GC。
如果当前节点的前置节点或者持有锁的动作失败了。则执行shouldParkAfterFailedAcquire(p, 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;
}
第一次进来的时候，waitStatus初始值是0，会直接走到compareAndSetWaitStatus(pred, ws, Node.SIGNAL);会把当前节点的前置节点的状态设置为SIGNAL，意思需要唤醒下一个节点。返回false。
也就是说会进入第二次循环，过程同上。也就是说会第二次进入shouldParkAfterFailedAcquire，如果waitStatus没有被修改过则返回true。如果被修改过，有可能被取消ws > 0，因为只有取消的状态是大于0的，这个时候需要一直往当前节点的前置节点找，一直找到不是取消状态的节点。这里为什么不给waitStatus设置默认为SIGNAL呢，减少循环次数。博客园有个大神说过，这是jdk大神故意为之，这样做尽量多执行一次获取锁的动作，类似自旋。因为如果这里判断为true后，该线程多半会阻塞 ，阻塞是会引起线程上下文切换降低吞吐量，很有道理。接着看：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
LockSupport.park是jdk提供的unsafe方法，大意就是阻塞该线程。并返回线程的中断状态。当然前置节点的线程没有unlock之前是不会返回的啦。lock方法的阅读就到此为止了。下面接着看unlock代码。图就不画了，直接开始。
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
这边意思比较简单，如果当前线程释放锁成功，则唤醒队列头节点的后置节点，先看下tryRelease。
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
releases值是1，状态值减1为0的话说明锁已经被 完全释放，持有锁的线程设置为null。如果不为0说明锁被重入过，这也说明了。调用了多少次lock，就必须有多少次unlock对应，不然就没有完全释放锁。然后看下unparkSuccessor。
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.
*/
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);
}
先把头节点waitStatus设置为默认值。获取头结点的后置节点，如果后置节点为null或者状态为取消，然后从链表的尾部开始遍历，直到找到链表中第一个不是取消状态的节点。这里为什么不直接从头结点往下遍历呢？最后再取消阻塞该节点。这个节点又开始执行acquireQueued的逻辑啦。
最后说一下公平锁和非公平锁中的区别。看下公平锁的逻辑。
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
相对非公平锁多了hasQueuedPredecessors的判断，这个方法的大概意思就是队列中是否存在等待的节点。如果有的话，公平锁的逻辑就不能直接尝试持有锁，而是需要进入到队列中排队。
ReentrantLock的关键方法已经分析完了，虽然是锁，但是jdk大神的设计中无处不提现出乐观的思想，减少因为线程阻塞引起的上下文切换，这也是为什么之前说ReentrantLock比synchronized的性能要好的多，当然synchronized经过各种优化已经很好很好啦，同样是用了乐观锁来替代悲观的操作。
网友评论