今天我們來研究學(xué)習(xí)一下ReentrantLock類的相關(guān)原理，ReentrantLock的內(nèi)部使用AbstractQueuedSynchronizer實(shí)現(xiàn)線程鎖。除了ReentrantLock在java.util.concurrent包中還有很多類都依賴于這個(gè)類所提供隊(duì)列式同步器。
為了方便學(xué)習(xí)我們以ReentrantLock為例，來學(xué)習(xí)ReentrantLock和AbstractQueuedSynchronizer。
本節(jié)我們先學(xué)習(xí)ReentrantLock是如何加鎖的。ReentrantLock.lock()內(nèi)部調(diào)用sync.lock()
ReentrantLock的內(nèi)部有兩種實(shí)現(xiàn)，F(xiàn)airSync（公平鎖）和NonfairSync（非公平鎖）都繼承了Sync類，
而Sync類繼承了AbstractQueuedSynchronizer。
AbstractQueuedSynchronizer
AbstractQueuedSynchronizer里有四個(gè)可以重寫的方法，可以用它們實(shí)現(xiàn)不同的功能。例如ReentrantLock、CountDownLatch、Semaphore等。
//獨(dú)占模式下獲取許可
@Override
protected boolean tryAcquire(int arg) {
return super.tryAcquire(arg);
}
//獨(dú)占模式下釋放許可
@Override
protected boolean tryRelease(int arg) {
return super.tryRelease(arg);
}
//共享模式下獲取許可
@Override
protected int tryAcquireShared(int arg) {
return super.tryAcquireShared(arg);
}
//共享模式下釋放許可
@Override
protected boolean tryReleaseShared(int arg) {
return super.tryReleaseShared(arg);
}
lock操作
下面將以NonfairSync為例，講解ReentrantLock是如何加鎖。
AbstractQueuedSynchronizer中有一個(gè)屬性state，為0時(shí)表示沒有線程持有鎖，大于0時(shí)表示有鎖。ReentrantLock是可重入表，因此state用于重入計(jì)數(shù)。
public class ReentrantLock {
private Sync sync;
//加鎖
public void lock() {
sync.lock();
}
static final class NonfairSync extends Sync {
final void lock() {
//CAS方式更新state變量的值，從0更新到1成功表示獲取到鎖
if (compareAndSetState(0, 1))
//記錄獲得鎖的線程
setExclusiveOwnerThread(Thread.currentThread());
else
//其他線程已經(jīng)獲取鎖走acquire
acquire(1);
}
}
abstract static class Sync extends AbstractQueuedSynchronizer {
......
}
}
acquire
acquire函數(shù)的作用是獲取同一時(shí)間段內(nèi)只能被一個(gè)線程獲取的許可。
先調(diào)用tryAcquire嘗試獲取一次鎖，如果失敗則調(diào)用acquireQueued等待鎖。
public final void acquire(int arg) {
//tryAcquire嘗試獲取鎖
//但是多線程搶鎖可能當(dāng)前線程搶不到，走acquireQueued
//addWaiter為當(dāng)前線程創(chuàng)建一個(gè)WaitNode節(jié)點(diǎn)加入隊(duì)列（鏈表）
if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
selfInterrupt();
}
}
//嘗試獲取一次鎖
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
//state為0表示沒有人持有鎖，可以嘗試獲取鎖
if (c == 0) {
//CAS方式更新state變量的值，從0更新到1成功表示獲取到鎖
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;
}
addWaiter
AbstractQueuedSynchronizer中實(shí)現(xiàn)了鏈表，用于保存所有的等待鎖的線程調(diào)用addWaiter將當(dāng)前線程封裝成鏈表中的一個(gè)節(jié)點(diǎn)加入鏈表中。
//鏈表頭節(jié)點(diǎn)
private transient volatile Node head;
//鏈表尾節(jié)點(diǎn)
private transient volatile Node tail;
//addWaiter為當(dāng)前線程創(chuàng)建一個(gè)WaitNode節(jié)點(diǎn)加入隊(duì)列（鏈表）
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
//嘗試向鏈表加入節(jié)點(diǎn)
//第一次調(diào)用addWaiter時(shí)tail為null，直接走enq
Node pred = tail;
if (pred != null) {
node.prev = pred;
//CAS方式插入node
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
private Node enq(final Node node) {
//不停嘗試插入node
for (;;) {
Node t = tail;
if (t == null) {
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
acquireQueued
addWaiter成功后，調(diào)用acquireQueued等待其他線程釋放鎖后重新獲取鎖。
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//不停嘗試獲取鎖
for (;;) {
//獲取前驅(qū)節(jié)點(diǎn)
final Node p = node.predecessor();
//前驅(qū)節(jié)點(diǎn)是頭節(jié)點(diǎn)時(shí)tryAcquire
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (//檢查node狀態(tài)
shouldParkAfterFailedAcquire(p, node)
//阻塞線程
//如果parkAndCheckInterrupt==true時(shí)線程已經(jīng)中斷
&& parkAndCheckInterrupt()) {
interrupted = true;
}
}
} finally {
if (failed)
cancelAcquire(node);
}
}
park
如果其他線程持有鎖，那么讓當(dāng)前線程阻塞。等到其他線程釋放鎖時(shí)喚醒線程
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//檢查前驅(qū)節(jié)點(diǎn)waitStatus（初始化時(shí)值為0）
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//如果前繼的節(jié)點(diǎn)狀態(tài)為SIGNAL，表示當(dāng)前節(jié)點(diǎn)需要阻塞
return true;
if (ws > 0) {
//過濾掉所有cancelled的節(jié)點(diǎn)
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
private final boolean parkAndCheckInterrupt() {
//阻塞線程, 等待其他線程釋放鎖
LockSupport.park(this);
//返回false，表示線程沒有中斷
return Thread.interrupted();
}
unlock操作
//解鎖
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
//嘗試釋放一個(gè)許可
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
//喚醒下一個(gè)Node中的線程
unparkSuccessor(h);
return true;
}
return false;
}
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
//判斷持有鎖的線程是否是當(dāng)前線程
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
//用于重入鎖
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
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);
}
