老板讓只懂Java基本語法的我，基于AQS實(shí)現(xiàn)一個(gè)鎖
10 點(diǎn)整，我到了公司，又成為全組最后一個(gè)到的員工。
正準(zhǔn)備刷刷手機(jī)摸摸魚，看見老板神秘兮兮地走了過來。
老板：閃客呀，你寫個(gè)工具，基于 AQS 實(shí)現(xiàn)一個(gè)鎖，給咱們組其他開發(fā)用
我：哦好的
老板：你多久能搞好？
我：就是一個(gè)工具類是吧，嗯今天下午就給你吧
老板：嗯，那你抓緊時(shí)間搞吧，大家都等著用呢
我：哦好的
先寫個(gè)框架
關(guān)于鎖，我還算有一個(gè)模糊的認(rèn)識(shí)的，要讓使用者可以獲取鎖、釋放鎖，來實(shí)現(xiàn)多線程訪問時(shí)的安全性。于是我趕緊先把一個(gè)框架寫了出來。
// 給帥氣老板用的鎖
public class FlashLock {
// 釋放鎖
public void lock() {}
// 釋放鎖
public void unlock() {}
}
工具類已經(jīng)完成一半了，一想到全組的開發(fā)們下午就會(huì)這樣用到我的工具，我不禁笑出了聲音。
FlashLock flashLock = new FlashLock();
public void doSomeThing() {
// 獲取鎖，表示同一時(shí)間只允許一個(gè)線程執(zhí)行這個(gè)方法
flashLock.lock();
try {
...
} finally {
// 優(yōu)雅地在 finally 里釋放鎖
flashLock.unlock();
}
}
隨著同事們投來異樣的眼光，我回過神來。繼續(xù)想，我怎么在這倆方法里實(shí)現(xiàn)這種鎖的效果呢？腦子一片空白呀，誒不過老板剛剛說要基于 AQS，那肯定這個(gè)東西可以給我提供一些方便吧，于是我在百度百科搜了一下什么是 AQS
百度百科尚未收錄詞條 “AQS”
這老板水平也太次了，給我推薦個(gè)百科上都搜不到的東西... 只能搜搜百度了
額！這看起來還是個(gè) Java 面試的重點(diǎn)呢！真是錯(cuò)怪老板了。
我點(diǎn)了其中一篇，了解到 AQS 的全稱叫 AbstractQueuedSynchronizer（抽象的隊(duì)列式同步器），是一個(gè) JDK 源碼中的一個(gè)類。
嗨，搞了半天只是個(gè)類而已嘛，對(duì)我這種源碼在手天下我有的神級(jí)碼農(nóng)，還看什么文章呀，我迅速打開了 JDK1.8 源碼，找到了這個(gè)類。
我的天，一共 2316 行！我趕緊把所有注釋都去掉，發(fā)現(xiàn)還有 914 行。
由于下午就要交稿，我打消了不看注釋硬啃源碼的念頭，開始從頭看起了注釋...
2：使用 AQS 實(shí)現(xiàn)最簡(jiǎn)單的鎖
Provides a framework for implementing blocking locks and related synchronizers (semaphores, events, etc) that rely on first-in-first-out (FIFO) wait queues. This class is designed to be a useful basis for most kinds of synchronizers that rely on a single atomic int value to represent state. Subclasses must define the protected methods that change this state, and which define what that state means in terms of this object being acquired or released.
第一句話說這是個(gè)框架，之后說這個(gè)類是基于一個(gè)原子變量，這說的都是原理我先不管。
后面又說子類（Subclasses）必須實(shí)現(xiàn)一些改變狀態(tài)（change this state）和獲取釋放鎖（acquired or released）的方法。
哦！看來我需要用一個(gè)子類繼承它，然后實(shí)現(xiàn)它指定的一些方法，其他的事情這個(gè)父類都會(huì)幫我做好的。敏銳的我馬上察覺到，這用的模板方法這種設(shè)計(jì)模式，這是我最喜歡的設(shè)計(jì)模式了，因?yàn)橹恍枰x懂需要讓子類實(shí)現(xiàn)的模板方法的含義，即可以很好地使用這個(gè)類的強(qiáng)大功能。
于是我趕緊去找，有哪些這樣的模板方法，需要子類去實(shí)現(xiàn)，果然在注釋中發(fā)現(xiàn)了這樣一段話。
* To use this class as the basis of a synchronizer, redefine the
* following methods
*
* <li> {@link #tryAcquire}
* <li> {@link #tryRelease}
* <li> {@link #tryAcquireShared}
* <li> {@link #tryReleaseShared}
* <li> {@link #isHeldExclusively}
* </ul>
在源碼中找到這幾個(gè)類
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}
protected boolean isHeldExclusively() {
throw new UnsupportedOperationException();
}
一看清一色都是拋出異常我就放心了，這正是留給我們子類實(shí)現(xiàn)的模板方法呀，接下來就是我寫個(gè)類實(shí)現(xiàn)他們就好咯，可是怎么寫...
正想去百度，突然發(fā)現(xiàn)注釋中居然給出了一段 基于 AQS 的實(shí)現(xiàn)小 demo，還挺長(zhǎng)，我理解了它的意思，并且把我看不懂的都去掉了，寫出了很簡(jiǎn)潔的鎖
public class FlashLock {
// 獲取鎖（這回填好骨肉了）
public void lock() {
sync.acquire(1);
}
// 釋放鎖
public void unlock() {
sync.release(1);
}
private final Sync sync = new Sync();
// 這個(gè)內(nèi)部類就是繼承并實(shí)現(xiàn)了 AQS 但我這里只先實(shí)現(xiàn)兩個(gè)方法
private static class Sync extends AbstractQueuedSynchronizer {
@Override
public boolean tryAcquire(int acquires) {
// CAS 方式嘗試獲取鎖，成功返回true，失敗返回false
if (compareAndSetState(0, 1)) {
return true;
}
return false;
}
@Override
protected boolean tryRelease(int releases) {
// 釋放鎖，這里為什么不像上面那樣也是 CAS 操作呢？請(qǐng)讀者思考
setState(0);
return true;
}
}
}
lock 和 unlock 方法都實(shí)現(xiàn)了，我趕緊寫個(gè)經(jīng)典的測(cè)試代碼
// 可能發(fā)生線程安全問題的共享變量
private static long count = 0;
// 兩個(gè)線程并發(fā)對(duì) count++
public static void main(String[] args) throws Exception {
// 創(chuàng)建兩個(gè)線程，執(zhí)行add()操作
Thread th1 = new Thread(()-> add());
Thread th2 = new Thread(()-> add());
// 啟動(dòng)兩個(gè)線程
th1.start();
th2.start();
// 等待兩個(gè)線程執(zhí)行結(jié)束
th1.join();
th2.join();
// 這里應(yīng)該是 20000 就對(duì)了，說明鎖生效了
System.out.println(count);
}
// 我畫了一上午寫出來的鎖，哈哈
private static ExampleLock exampleLock = new ExampleLock();
// 循環(huán) count++，進(jìn)行 10000 次
private static void add() {
exampleLock.lock();
for (int i = 0; i < 10000; i++) {
count++;
}
add2();
// 沒啥異常，我就直接釋放鎖了
exampleLock.unlock();
}
測(cè)了好幾次，發(fā)現(xiàn)都是 20000，哈哈，大功告成，我趕緊在大群里 @所有人，告訴大家我寫的新工具。同事和老板紛紛給我點(diǎn)了贊。
我又忍不住笑出了聲音。走出了公司，準(zhǔn)備找個(gè)地方吃午飯。
不得不研究下 AQS 的原理
下午兩點(diǎn)整，我又成為公司最后一個(gè)午睡起床的人...
小宇：閃客，你的工具類確實(shí)好用，而且源碼也很簡(jiǎn)潔
我：哈哈，大家喜歡用就好
小宇：不過我有個(gè)問題，就是我用你的這個(gè)鎖工具，有的線程總是搶不到鎖，有的線程總是能搶到鎖。雖說線程們搶鎖確實(shí)看命，但能不能加入一種設(shè)計(jì)，讓各個(gè)線程機(jī)會(huì)均等些，起碼不要出現(xiàn)某幾個(gè)線程總是特倒霉搶不到鎖的情況呢？
我：這怎么可能，我就是寫個(gè)鎖工具，還能影響到人家 CPU 和操作系統(tǒng)層面的機(jī)制？
小宇：你想想吧，作為公司最帥的程序猿，我相信你哦
我：額這...
我這人最禁不住妹子夸獎(jiǎng)，趕緊開啟電腦屏幕，盯著我的獲取鎖的代碼看
@Override
public boolean tryAcquire(int acquires) {
// 一上來就 CAS 搶鎖
if (compareAndSetState(0, acquires)) {
return true;
}
return false;
}
我發(fā)現(xiàn)這段代碼中在嘗試獲取鎖時(shí)，一上來就 CAS 搶鎖，一旦成功就返回了 true。那我這里是否能加入某些機(jī)制，使這些線程不要一有機(jī)會(huì)就開始直接開始搶鎖，而是先考慮一下其他線程的感受再?zèng)Q定是否搶鎖呢？
我發(fā)現(xiàn)此時(shí)不得不研究一下 AQS 的內(nèi)部實(shí)現(xiàn)邏輯了，也就是原理，看看能不能得到一些思路。
我看 AQS 雖然方法一大堆，但屬性一共就四個(gè)（有一個(gè)是內(nèi)部類 Node）
public abstract class AbstractQueuedSynchronizer {
private transient volatile Node head;
private transient volatile Node tail;
private volatile int state;
static final class Node {}
}
static final class Node {
// ... 省略一些暫不關(guān)注的
volatile Node prev;
volatile Node next;
volatile Thread thread;
}
結(jié)合最開始看那段對(duì) AQS 高度概括的注釋
Provides a framework for implementing blocking locks and related synchronizers (semaphores, events, etc) that rely on first-in-first-out (FIFO) wait queues. This class is designed to be a useful basis for most kinds of synchronizers that rely on a single atomic int value to represent state.
不難猜到這里的內(nèi)部類 Node 以及其類型的變量 head 和 tail 就表示 AQS 內(nèi)部的一個(gè)等待隊(duì)列，而剩下的 state 變量就用來表示鎖的狀態(tài)。
等待隊(duì)列應(yīng)該就是線程獲取鎖失敗時(shí)，需要臨時(shí)存放的一個(gè)地方，用來等待被喚醒并嘗試獲取鎖。再看 Node 的屬性我們知道，Node 存放了當(dāng)前線程的指針 thread，也即可以表示當(dāng)前線程并對(duì)其進(jìn)行某些操作，prev 和 next 說明它構(gòu)成了一個(gè)雙向鏈表，也就是為某些需要得到前驅(qū)或后繼節(jié)點(diǎn)的算法提供便利。
太好了，僅僅看一些屬性和一段注釋，就得到了一個(gè)關(guān)于 AQS 大致原理的猜測(cè)，看起來還挺靠譜，我趕緊把它畫成幾張圖來加深理解。（由于這里非常重要，就不再賣關(guān)子了，直接畫出最正確的圖理解，但不會(huì)過于深入細(xì)節(jié)）
以下的圖是 AQS 最為核心的幾行代碼的直觀理解過程，請(qǐng)大家仔細(xì)品味
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
看完圖沒太消化的，這里還有一次機(jī)會(huì)，我來捋一捋。
首先初始狀態(tài) AQS 的 state=0，表示沒有線程持有鎖。head 和 tail 也都為空，表示此時(shí)等待隊(duì)列里也沒有線程。
這時(shí)第一個(gè)線程（線程1）來了，沒有任何線程和它搶，直接拿到了鎖（state=1）
然后線程2也來了，假設(shè)此時(shí)線程1沒有釋放鎖，那么線程2搶鎖失敗（執(zhí)行你自己寫的 tryAcquire）。失敗后，剩下的事都是 AQS 幫你做的，首先加入等待隊(duì)列的隊(duì)尾 addWaiter?（此時(shí)隊(duì)列為空，所以要先初始化一個(gè)占位的頭結(jié)點(diǎn)），然后在循環(huán)里嘗試獲取鎖 acquireQueue（注意這里面有相當(dāng)多的細(xì)節(jié)，首先前驅(qū)結(jié)點(diǎn)是頭結(jié)點(diǎn)的才能嘗試獲取鎖，即排在隊(duì)頭的才有機(jī)會(huì)。再有循環(huán)里獲取鎖并不是一直 CAS，而是通過一個(gè)標(biāo)志控制了次數(shù)，使得 CAS 兩次都失敗后就將線程掛起 park，之后等待持有鎖的線程釋放鎖之后再喚醒 unpark。其余各種細(xì)節(jié)，希望讀者閱讀源碼，不是一句兩句說清楚的）。
然后線程3也來了?，經(jīng)歷了和線程2一樣的經(jīng)歷，只不過它的前驅(qū)結(jié)點(diǎn)不是頭結(jié)點(diǎn)，因此還不能有機(jī)會(huì)嘗試獲取鎖，只有等線程2搶到了鎖并且出隊(duì)，自己的前驅(qū)結(jié)點(diǎn)變成了頭結(jié)點(diǎn)，才可以。
這時(shí)線程1終于釋放了鎖（state=0），與此同時(shí)找到隊(duì)列的頭結(jié)點(diǎn)進(jìn)行喚醒 unpark。此時(shí)頭結(jié)點(diǎn)是線程2表示的 Node，因此對(duì)線程2進(jìn)行了喚醒操作。如果此時(shí)線程2沒有被掛起，說明還在嘗試獲取鎖的過程中，那么就嘗試好了。如果已經(jīng)被掛起了，那么喚醒線程2，使得線程2繼續(xù)不斷嘗試 CAS 獲取鎖，直到成功為止。
如此，循環(huán)往復(fù)... 你大概懂了么？
嗯原理搞懂了，實(shí)現(xiàn)一個(gè)公平鎖
仔細(xì)看上面的倒數(shù)第二張圖。
好好好，你懶得往上翻，我給你粘過來。
原本在隊(duì)列中等待的線程 2，被線程 1 釋放鎖之后喚醒了，但它仍然需要搶鎖，而且有可能搶失敗。
那如果每次這個(gè)線程 2 嘗試搶鎖時(shí)，都有其他新來的線程把鎖搶去，那線程 2 就一直得不到運(yùn)行機(jī)會(huì)，而且排在線程 2 后面的等待線程，也都沒有機(jī)會(huì)運(yùn)行。
導(dǎo)致有的線程一直得不到運(yùn)行機(jī)會(huì)的，就是這個(gè)新進(jìn)來的線程每次都不管有沒有人排隊(duì)，都直接上來就搶鎖導(dǎo)致的。
妥了，剛剛小宇提出的問題，我終于有了思路，就是讓新來的線程搶鎖時(shí)，先問一句，“有沒有人排隊(duì)呀？如果有人排隊(duì)那我先排到隊(duì)尾好了”。
@Override
public boolean tryAcquire(int acquires) {
// 原有基礎(chǔ)上加上這個(gè)
if (有線程在等待隊(duì)列中) {
// 返回獲取鎖失敗，AQS會(huì)幫我把該線程放在等待隊(duì)列隊(duì)尾的
return false;
}
if (compareAndSetState(0, 1)) {
return true;
}
return false;
}
怎么判斷是否有線程在等待隊(duì)列呢？機(jī)智的我覺得，AQS 這么優(yōu)秀的框架一定為上層提供了一個(gè)方法，不會(huì)讓我們深入到它實(shí)現(xiàn)的內(nèi)部的，果然我找到了。
public final boolean hasQueuedPredecessors()
再經(jīng)過優(yōu)化結(jié)構(gòu)后，最終的代碼變成了這樣
@Override
public boolean tryAcquire(int acquires) {
if (hasQueuedPredecessors() &&
compareAndSetState(0, 1)) {
return true;
}
return false;
}
哈哈，大功告成，趕緊去找小宇顯擺一下。
等等...
那我原來的那種實(shí)現(xiàn)方式就沒了，肯定有其他人找我質(zhì)問，emmm，我兩種方式都暴露給大家吧，隨大家選。
我將原來的暴力搶鎖方式起了個(gè)名，叫非公平鎖，因?yàn)榫€程搶鎖不排隊(duì)，純看臉。按小宇需求實(shí)現(xiàn)的排隊(duì)獲取鎖，我叫它公平鎖，因?yàn)橹灰芯€程在排隊(duì)，新來的就得乖乖去排隊(duì)，不能直接搶。
// 想要公平鎖，就傳 true 進(jìn)來
public FlashLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
哈哈，有了高大上的名字和代碼實(shí)現(xiàn)，我興高采烈去找小宇交差了。
老板要求方法可以重入
晚上五點(diǎn)半，我正準(zhǔn)備成為全組第一個(gè)去吃飯的人，突然老板陰著臉跑了過來。
老板：閃客，我用你這工具導(dǎo)致一個(gè)線程卡死了呀，一直獲取不到鎖
我：嗯怎么會(huì)呢？
老板：代碼發(fā)你了，你趕緊看看！
我打開了鎖了屏的電腦，點(diǎn)開了老板發(fā)來的代碼
public void doSomeThing2() {
flashLock.lock();
doSomeThing2();
flashLock.unlock();
}
public void doSomeThing2() {
flashLock.lock();
...
flashLock.unlock();
}
我恍然大悟，原來一個(gè)線程執(zhí)行了一個(gè)方法，獲取了鎖，這個(gè)方法沒有結(jié)束，又調(diào)用了另一個(gè)需要鎖的方法，于是卡在這再也不走了。
這個(gè)原理很容易理解，但這似乎用起來確實(shí)不太友好，怪不得老板那么生氣。有沒有辦法，讓同一個(gè)線程持有鎖時(shí)，還能繼續(xù)獲取鎖（可重入），只有當(dāng)不同線程才互斥呢？
我苦思冥想，感覺不對(duì)呀，現(xiàn)在 AQS 里面的所有變量我都用到了，沒見哪個(gè)變量可以記錄當(dāng)前線程呀。
哦對(duì)！AQS 本身還繼承了 AbstractOwnableSynchronizer 這個(gè)類！我很快在這個(gè)類里面發(fā)現(xiàn)了這個(gè)屬性！
/**
* The current owner of exclusive mode synchronization.
*/
private transient Thread exclusiveOwnerThread;
熟悉了之前的套路，我很快又找到了這兩個(gè)方法！
protected final void setExclusiveOwnerThread(Thread thread);
protected final Thread getExclusiveOwnerThread();
大功告成，此時(shí)我只要在一個(gè)線程發(fā)現(xiàn)鎖已經(jīng)被占用時(shí)，不直接放棄，而是再看一下占用鎖的線程是不是正是我自己，就好了。有了前面的經(jīng)驗(yàn)，這次我直接寫出了最終的可重入的公平鎖代碼。
@Override
public boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (hasQueuedPredecessors() && compareAndSetState(0, 1)) {
// 拿到鎖記得記錄下持鎖線程是自己
setExclusiveOwnerThread(current);
return true;
}
} else if (current == getExclusiveOwnerThread()) {
// 看見鎖被占了(state=0)也別放棄，看看是不是自己占的
setState(c + acquires);
return true;
}
return false;
}
6. 下班！
我把這個(gè)最終版的鎖代碼提交，霸氣地收拾東西下班了，今天真是收獲滿滿。
好啦故事講完了，如果你堅(jiān)持讀到了這里并且完全理解了上面的所有事情，那么恭喜你，你已經(jīng)掌握了 AQS 的核心原理以及基于它的一個(gè)經(jīng)典的鎖實(shí)現(xiàn) ReentrantLock 的幾乎全部知識(shí)點(diǎn)，AQS 的體系骨架算是被你不知不覺建立起來了，這兩個(gè)都是 Java 程序員面試必備的東西。
雖然這只是皮毛，但如果你是第一次接觸這兩個(gè)概念，那本篇文章的最大意義在于對(duì)他們有了一個(gè)三觀很正的第一印象。我希望 AQS 的給你的第一印象不是什么抽象的隊(duì)列式同步器，而只是一個(gè)為了更方便實(shí)現(xiàn)各種鎖而提供的一個(gè)包含幾個(gè)模板方法的類而已，雖然并不準(zhǔn)確，而且顯得很 low，但實(shí)則可能恰恰是說到了本質(zhì)。
7. 繼續(xù)深入 AQS
我之后也會(huì)出關(guān)于 AQS 繼續(xù)深入的文章，不過下面的三篇系列文章你可以花上兩三個(gè)小時(shí)在電腦上看一下，真的非常非常非常給力。
另外，我也推薦你，用跟蹤源碼或 debug 的方式，從頭到尾自己跟一遍下面三行代碼，是幾乎 AQS 的全部核心邏輯，這個(gè)看懂了，其他的都是浮云。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
最后呢，如果你這兩個(gè)都看不下去，關(guān)注低并發(fā)編程，同樣能得到有趣且深入的理解，哈哈哈。
公眾號(hào) - 低并發(fā)編程
浙公網(wǎng)安備 33010602011771號(hào)