AQS實(shí)現(xiàn)原理
AQS中維護(hù)了一個(gè)volatile int state（共享資源）和一個(gè)CLH隊(duì)列。當(dāng)state=1時(shí)代表當(dāng)前對(duì)象鎖已經(jīng)被占用，其他線程來(lái)加鎖時(shí)則會(huì)失敗，失敗的線程被放入一個(gè)FIFO的等待隊(duì)列中，然后會(huì)被UNSAFE.park()操作掛起，等待已經(jīng)獲得鎖的線程釋放鎖才能被喚醒。
我們拿具體場(chǎng)景來(lái)分析，假設(shè)同時(shí)有三個(gè)線程并發(fā)搶占鎖，此時(shí)線程一搶占成功，線程二、三搶占失敗，具體流程如下：
此時(shí)AQS內(nèi)部數(shù)據(jù)結(jié)構(gòu)為：
上圖可以看到等待隊(duì)列中的節(jié)點(diǎn)Node是一個(gè)雙向鏈表，這里SIGNAL是Node中waitStatus屬性。
以非公平鎖看下具體實(shí)現(xiàn)：
java.util.concurrent.locks.ReentrantLock.NonfairSync：
static final class NonfairSync extends Sync {
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
線程進(jìn)來(lái)直接利用CAS嘗試搶占鎖，如果搶占成功state值會(huì)被修改為1，且設(shè)置對(duì)象獨(dú)占鎖線程為當(dāng)前線程。
線程搶占實(shí)現(xiàn)#
線程二搶占失敗，執(zhí)行acquire(1)方法。
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire()：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire是AbstractQueuedSynchronizer的方法，未提供對(duì)應(yīng)實(shí)現(xiàn)，由子類實(shí)現(xiàn)：
java.util.concurrent.locks.ReentrantLock .nonfairTryAcquire()：
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)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
nonfairTryAcquire()方法中首先會(huì)獲取state的值，如果不為0則說(shuō)明當(dāng)前對(duì)象的鎖已經(jīng)被其他線程占有，接著判斷占有鎖的線程是否為當(dāng)前線程，如果是則累加state值，這里其實(shí)就是可重入鎖的具體實(shí)現(xiàn)。如果state為0，則執(zhí)行CAS操作，嘗試更新state值為1，如果更新成功則代表當(dāng)前線程加鎖成功。
當(dāng)前線程二執(zhí)行tryAcquire()后返回false，接著執(zhí)行addWaiter(Node.EXCLUSIVE)邏輯，將自己加入到一個(gè)FIFO等待隊(duì)列中，代碼實(shí)現(xiàn)如下：
java.util.concurrent.locks.AbstractQueuedSynchronizer.addWaiter()：
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
此時(shí)隊(duì)列中tail指針為空，直接調(diào)用enq(node)方法將當(dāng)前線程加入等待隊(duì)列尾部：
private Node enq(final 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;
}
}
}
}
第一次循環(huán)時(shí)tail為空，創(chuàng)建一個(gè)哨兵節(jié)點(diǎn)，head指向這個(gè)哨兵節(jié)點(diǎn)；第二次循環(huán)，將線程二對(duì)應(yīng)的node節(jié)點(diǎn)掛載到head節(jié)點(diǎn)后面并返回當(dāng)前線程創(chuàng)建的節(jié)點(diǎn)信息。繼續(xù)往后執(zhí)行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)邏輯，此時(shí)傳入的參數(shù)為線程二對(duì)應(yīng)的node節(jié)點(diǎn)信息。
java.util.concurrent.locks.AbstractQueuedSynchronizer.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) &&
parkAndChecknIterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
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);
return Thread.interrupted();
}
acquireQueued()會(huì)先判斷當(dāng)前傳入的Node對(duì)應(yīng)的前置節(jié)點(diǎn)是否為head，如果是則嘗試加鎖。加鎖成功則將當(dāng)前節(jié)點(diǎn)設(shè)置為head節(jié)點(diǎn)，然后刪除之前的head節(jié)點(diǎn)。
如果加鎖失敗或者Node的前置節(jié)點(diǎn)不是head節(jié)點(diǎn)，就會(huì)通過(guò)shouldParkAfterFailedAcquire方法將head節(jié)點(diǎn)的waitStatus變成SIGNAL=-1，最后執(zhí)行parkAndChecknIterrupt方法，調(diào)用LockSupport.park()掛起當(dāng)前線程。此時(shí)線程二需要等待其他線程釋放鎖來(lái)喚醒。
線程釋放實(shí)現(xiàn)#
線程一執(zhí)行完后釋放鎖，具體代碼如下：
java.util.concurrent.locks.AbstractQueuedSynchronizer.release()：
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
先執(zhí)行tryRelease方法，如果執(zhí)行成功，則繼續(xù)判斷head節(jié)點(diǎn)的waitStatus是否為0，這個(gè)值為SIGNAL=-1不為0，繼續(xù)執(zhí)行unparkSuccessor()方法喚醒head的后置節(jié)點(diǎn)。
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);
}
setState(c);
return free;
}
執(zhí)行完ReentrantLock.tryRelease()后，state被設(shè)置為0，Lock對(duì)象的獨(dú)占鎖被設(shè)置為null。
接著執(zhí)行java.util.concurrent.locks.AbstractQueuedSynchronizer.unparkSuccessor()方法，喚醒head的后置節(jié)點(diǎn)：
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);
}
這里主要是將head節(jié)點(diǎn)的waitStatus設(shè)置為0，然后解除head節(jié)點(diǎn)next的指向，使head幾點(diǎn)空置，等待被垃圾回收。
此時(shí)重新將head指針指向線程二對(duì)應(yīng)的Node節(jié)點(diǎn)，且使用LockSupport.unpark方法來(lái)喚醒線程二。被喚醒的線程會(huì)接著嘗試獲取鎖，用CAS指令修改state數(shù)據(jù)。執(zhí)行完成后AQS中的數(shù)據(jù)結(jié)構(gòu)如下：
