要理解ReentrantLock，首先要理解AbstractQueuedSynchronizer。
AbstractQueuedSynchronizer类如其名，抽象的队列式的同步器，AQS定义了一套多线程访问共享资源的同步器框架，许多同步类实现都依赖于它，如常用的ReentrantLock/Semaphore/CountDownLatch。
image
它维护了一个volatile int state（代表共享资源）和一个FIFO线程等待队列（多线程争用资源被阻塞时会进入此队列）。
acquire方法表示申请资源，它调用tryAcquire方法（tryAcquire由子类实现），如果tryAcquire失败，会将当前线程转化为等待队列的一个节点。
release方法表示释放资源，它调用tryRelease（tryRelease由子类实现），如果tryRelease成功，会唤醒等待队列某个节点的线程。
除了tryAcquire/tryRelease，AbstractQueuedSynchronizer还提供了tryAcquireShared/tryReleaseShared/ isHeldExclusively。
不同的同步类正是通过这些方法实现不同逻辑。
ReentrantLock实现了公平锁FairSync和非公平锁NonfairSync，它们继续自ReentrantLock内部抽象类Sync，Sync实现了一些公用的逻辑。
这篇文章只关注NonfairSync。
单线程加锁
查看NonfairSync.lock
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
compareAndSetState是AbstractQueuedSynchronizer提供的方法，通过CAS争夺AbstractQueuedSynchronizer的资源state。
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
单线程环境下，compareAndSetState(0, 1)可以成功，setExclusiveOwnerThread设置持有锁的线程为当前线程。
否则调用AbstractQueuedSynchronizer.acquire重新争夺资源。
单线程解锁
unlock调用AbstractQueuedSynchronizer.release，它会先调用ReentrantLock.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); // 持有锁的线程设置为null
}
setState(c); // 更新State
return free;
}
单线程环境下，只需要将state减去releases的值，State为0时，清除当前持锁线程就行。
setState没有使用CAS操作，因为当前线程已经占有资源(state)，注意：state是volatile的。
多线程环境，可能还要唤醒等待队列中的线程，后面再说。
多线程抢锁
抢锁的关键在NonfairSync.lock方法中的compareAndSetState(0, 1)操作，CAS保证只有一个线程抢锁成功，其他线程将调用AbstractQueuedSynchronizer.acquire
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire是AbstractQueuedSynchronizer提供的抽象方法，实现在Sync.nonfairTryAcquire
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) { // state == 0,重新抢锁
if (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;
}
如果state不是0而且持有锁的非当前线程，返回false。AbstractQueuedSynchronizer就将当前线程转化为一个Waiter，添加到等待队列中。
回到addWaiter(Node.EXCLUSIVE), arg)，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;
}
enq方法将node插入等待队列，如果head/tail为null会初始化他们。
如果compareAndSetTail(pred, node)的CAS操作失败(多个线程操作)，也会进入enq方法。
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // 初始化tail/head
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) { // 更新tail
t.next = node;
return t;
}
}
}
}
可以看到，head是一个标识节点，它不代表一个请求锁的线程。
这里for (;;)可以保证初始化head/tail或更新tail操作成功，否则不断重试。
addWaiter完成，回来看看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)) { // 前一个节点是head，重新抢锁
setHead(node); // 抢锁成功，设置head为node
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
如果某个节点抢锁成功，会将这个节点设置为head。
如果抢锁失败，调用shouldParkAfterFailedAcquire
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
if (ws > 0) { // ws = CANCELLED
// CODE1
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// CODE2
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
waitStatus是一个等待队列节点的内部状态，有
SIGNAL：该节点的后继节点（或将）被阻塞，因此当前节点释放或取消时必须唤醒其后继节点。
CANCELLED：由于超时或中断，此节点被取消。
CONDITION/PROPAGATE暂时不关注。
0是默认状态。
只有CANCELLED状态大于0，pred.waitStatus > 0表示pred为已取消节点。所以CODE1处理会清除当前节点前的CANCELLED状态节点。
CODE2会更改prev节点的状态为SIGNAL。
如果pred.waitStatus不是SIGNAL，shouldParkAfterFailedAcquire会在acquireQueued的for循环中重复调用，直到返回true。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
LockSupport.park(this);会挂起当前线程，阻塞等待。
这里返回并清除线程的中断信息，最后acquireQueued方法将线程中断信息interrupted返回。
回到acquire方法，如果acquireQueued返回true，将中断当前线程。
多线程解锁
ReentrantLock.unlock调用AbstractQueuedSynchronizer.方法
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
head状态不能为CANCELLED，
head不为null而且waitStatus不是初始状态0时，调用unparkSuccessor，唤醒head的下一个节点线程
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0) // 清除当前节点waitStatus
compareAndSetWaitStatus(node, ws, 0);
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);
}
如果s为null或s.waitStatus为CANCELLED，则从tail向前遍历，找到最前面一个waitStatus不为CANCELLED的节点。
LockSupport.unpark(s.thread)唤醒LockSupport.park挂起的线程。
这时前面阻塞的线程回到acquireQueued方法，重新执行for (;;)循环，重新tryAcquire。
如果其他线程此时也调用lock方法，可能其他线程加锁成功，被唤醒的线程重新阻塞(抢锁的关键在于compareAndSetState(0, 1))。
tryLock
tryLock调用到AbstractQueuedSynchronizer.tryAcquireNanos
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted()) // 获取并清除中断信息
throw new InterruptedException();
return tryAcquire(arg) ||
doAcquireNanos(arg, nanosTimeout);
}
同样，tryAcquire失败后，调用doAcquireNanos:
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout; // 超时时间
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) { // 重新抢锁
setHead(node);
p.next = null; // help GC
failed = false;
return true;
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L) // 已超时
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout); // 超时阻塞
if (Thread.interrupted()) // 已中断
throw new InterruptedException();
}
} finally {
// 超时，撤销操作
if (failed)
cancelAcquire(node);
}
}
和acquireQueued方法类似。
当nanosTimeout > spinForTimeoutThreshold时，不会挂起线程，而是不断tryAcquire，这里是自旋锁。
spinForTimeoutThreshold是1000纳秒。
cancelAcquire
CANCELLED状态在cancelAcquire方法中产生，如果加锁异常，中断或超时会调用这个方法
private void cancelAcquire(Node node) {
if (node == null)
return;
node.thread = null;
// 跳过前面CANCELLED的节点
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
Node predNext = pred.next;
// 修改waitStatus为CANCELLED
node.waitStatus = Node.CANCELLED;
// 如果是tail，移除该节点
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
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)
// 修改next-link
compareAndSetNext(pred, predNext, next);
} else {
unparkSuccessor(node); // 唤醒下一个节点
}
node.next = node; // help GC
}
}
如果pred节点是head（本来该当前节点运行了），直接唤醒当前节点的下一个节点。
否则通过CAS操作更新pred节点的WaitStatus为SIGNAL，成功则修改pred节点的next属性指向下一个节点，失败（可能pred节点已经变化了），唤醒当前节点的下一个节点。
可中断锁
lockInterruptibly方法调用AbstractQueuedSynchronizer.doAcquireInterruptibly，这个方法和acquireQueued差不多
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException(); // 抛出异常，中断锁
}
} finally {
if (failed)
cancelAcquire(node);
}
}
注意：LockSupport.park是可以响应中断的。
网友评论