摘要：但是不管怎樣，在一個線程已經獲取鎖后，在釋放前再次獲取鎖是一個合理的需求，而且并不生硬。那么如果考慮重入，也很簡單，在加鎖時將的值累加即可，表示同一個線程重入此鎖的次數，當歸零，即表示釋放完畢。

最近研究了一下juc包的源碼。
在研究ReentrantReadWriteLock讀寫鎖的時候，對于其中一些細節的思考和處理以及關于提升效率的設計感到折服，難以遏制想要分享這份心得的念頭，因此在這里寫一篇小文章作為記錄。

本片文章建立在已經了解并發相關基礎概念的基礎上，可能不會涉及很多源碼，以思路為主。
如果文章有什么紕漏或者錯誤，還請務必指正，預謝。

首先我們需要知道獨占鎖（RenentractLock）這種基礎的鎖，在juc中是如何實現的：
它基于java.util.concurrent.locks.AbstractQueuedSynchronizer（AQS）這個抽象類所搭建的同步框架，利用AQS中的一個volatile int類型變量的CAS操作來表示鎖的占用情況，以及一個雙向鏈表的數據結構來存儲排隊中的線程。

簡單地說：一個線程如果想要獲取鎖，就需要嘗試對AQS中的這個volatile int變量（下面簡稱state）執行類似comapre 0 and swap 1的操作，如果不成功就進入同步隊列排隊（自旋和重入之類的細節不展開說了）同時休眠自己（LockSupport.park），等待占有鎖的線程釋放鎖時再喚醒它。

那么，如果不考慮重入，state就是一個簡單的狀態標識：0表示鎖未被占用，1表示鎖被占用，同步性由volatile和CAS來保證。

上面說的是獨占鎖，state可以不嚴謹地認為只有兩個狀態位。

但是如果是讀寫鎖，那這個鎖的基本邏輯應該是：讀和讀共享、讀和寫互斥、寫和寫互斥

如何實現鎖的共享呢？
如果我們不再把state當做一個狀態，而是當做一個計數器，那仿佛就可以說得通了：獲取鎖時compare n and swap n+1，釋放鎖時compare n and swap n-1，這樣就可以讓鎖不被獨占了。

因此，要實現讀寫鎖，我們可能需要兩個鎖，一個共享鎖（讀鎖），一個獨占鎖（寫鎖），而且這兩個鎖還需要協作，寫鎖需要知道讀鎖的占用情況，讀鎖需要知道寫鎖的占用情況。

設想中的簡單流程大概如下：

設想中的流程很簡單，然而存在一些問題：

關于讀寫互斥：

對于某個線程，當它先獲得讀鎖，然后在執行代碼的過程中，又想獲得寫鎖，這種情況是否應該允許？

如果允許，會有什么問題？如果不允許，當真的存在這種需求時怎么辦？

關于寫寫互斥是否也存在上面兩條提到的情況和問題呢？

我們知道一般而言，讀的操作數量要遠遠大于寫的操作，那么很有可能讀鎖一旦被獲取就長時間處于被占有的情況（因為新來的讀操作只需要進去+1就好了，不需要等待state回到0，這樣的話state可能永遠不會有回到0的一天），這會導致極端情況下寫鎖根本沒有機會上位，該如何解決這種情況？

對于上面用計數器來實現共享鎖的假設，當任意一個線程想要釋放鎖（即使它并未獲取鎖，因為解鎖的方法是開放的，任何獲取鎖對象的線程都可以執行lock.unlock()）時，如何判斷它是否有權限執行compare n and swap n-1？

是否應該使用ThreadLocal來實現這種權限控制？

如果使用ThreadLocal來控制，如何保證性能和效率？

在開始研究ReentrantReadWriteLock之前，應當先了解兩個概念：

重入性
一個很現實的問題是：我們時常需要在鎖中加鎖。這可能是由代碼復用產生的需求，也可能業務的邏輯就是這樣。
但是不管怎樣，在一個線程已經獲取鎖后，在釋放前再次獲取鎖是一個合理的需求，而且并不生硬。
上文在說獨占鎖時說到如果不考慮重入的情況，state會像boolean一樣只有兩個狀態位。那么如果考慮重入，也很簡單，在加鎖時將state的值累加即可，表示同一個線程重入此鎖的次數，當state歸零，即表示釋放完畢。

公平、非公平
這里的公平和非公平是指線程在獲取鎖時的機會是否公平。
我們知道AQS中有一個FIFO的線程排隊隊列，那么如果所有線程想要獲取鎖時都來排隊，大家先來后到井然有序，這就是公平；而如果每個線程都不守秩序，選擇插隊，而且還插隊成功了，那這就是不公平。
但是為什么需要不公平呢？
因為效率。
有兩個因素會制約公平機制下的效率：

上下文切換帶來的消耗

依賴同步隊列造成的消耗

我們之所以會使用鎖、使用并發，可能很大一部分原因是想要挖掘程序的效率，那么相應的，對于性能和效率的影響需要更加敏感。
簡單地說，上述的兩點由于公平帶來的性能損耗很可能讓你的并發失去高效的初衷。
當然這也是和場景密切關聯的，比如說你非常需要避免ABA問題，那么公平模式很適合你。
具體的不再展開，可以參考這篇文章：深入剖析ReentrantLock公平鎖與非公平鎖源碼實現

回到我們之前提的問題：

我們先考慮這種情況是否實際存在：假設我們有一個對象，它有兩個實例變量a和b，我們需要在實現：if a = 1 then set b = 2，或者換個例子，如果有個用戶名字叫張三就給他打錢。

這看上去仿佛是個CAS操作，然而它并不是，因為它涉及了兩個變量，CAS操作并不支持這種針對多個變量的疑似CAS的操作。
為什么不支持呢？因為cpu不提供這種針對多個變量的CAS操作指令（至少x86不提供），代碼層面的CAS只是對cpu指令的封裝而已。
為什么cpu不支持呢？可以，但沒必要鄙人也不是特別清楚（逃）。

總而言之這種情況是存在的，但是在并發情況下如果不加鎖就會有問題：比如先判斷得到這個用戶確實名叫張三，正準備打錢，突然中途有人把他的名字改了，那再打這筆錢就有問題了，我們判斷名字和打錢這兩個行為中間應當是沒有空隙的。
那么為了保證這個操作的正確性，我們或許可以在讀之前加一個讀鎖，在我釋放鎖之前，其他人不得改變任何內容，這就是之前所說的讀寫互斥：讀的期間不準寫。
但是如果照著這個想法，就產生了自相矛盾的地方：都說了讀期間不能寫，那你自己怎么在寫（打錢）呢？

如果我們順著這個思路去嘗試解釋“自己讀的期間還在寫”的行為的正當性，我們也許可以設立一個規則：如果讀鎖是我自己持有，則我可以寫。
然而這會出現其他的問題：因為讀鎖是共享的，你加了讀鎖，其他人仍然可以讀，這是否會有問題呢？假如我們的打錢操作涉及更多的值的改變，只有這些值全部改變完畢，才能說此時的整體狀態正確，否則在改變完畢之前，讀到的東西都有可能是錯的。
再去延伸這個思路似乎會變得非常艱難，也許會陷入耦合的地獄。

但是實際上我們不需要這樣做，我們只需要反過來使用讀寫互斥的概念即可：因為寫寫互斥（寫鎖是獨占鎖），所以我們在執行這個先讀后寫的行為之前，加一個寫鎖，這同樣能防止其他人來寫，同時還可以阻止其他人來讀，從而實現我們在單線程中讀寫并存的需求。

這就是ReentrantReadWriteLock中一個重要的概念：鎖降級

對于另一個子問題：如果在已經獲取寫鎖的期間還要再獲取寫鎖的時候怎么辦？
這種情況還是很常見的，多數是由于代碼的復用導致，不過相應的處理也很簡單：對寫鎖這個獨占鎖增加允許單線程重入的規則即可。

如果我們有兩把鎖，一把讀鎖，一把寫鎖，它們之間想要互通各自加鎖的情況很簡單——只要去get對方的state就行了。
但是只知道state是不夠的，對于讀的操作來說，它如果只看到寫鎖沒被占用，也不管有多少個寫操作還在排隊，就去在讀鎖上+1，那很可能發展成為問題所說的場景：寫操作永遠沒機會上位。

那么我們理想的情況應該是：讀操作如果發現寫鎖空閑，最好再看看寫操作的排隊情況如何，酌情考慮放棄這一次競爭，讓寫操作有機會上位。
這也是我理解的，為什么ReentrantReadWriteLock不設計成兩個互相溝通的、獨立的鎖，而是公用一個鎖（class Sync extends AbstractQueuedSynchronizer）——因為它們看似獨立，實際上對于耦合的需求很大，它們不僅需要溝通鎖的情況，還要溝通隊列的情況。

公用一個鎖的具體實現是：使用int state的高16位表示讀鎖的state，低16位表示寫鎖的state，而隊列公用的方式是給每個節點增加一個標記，表明該節點是一個共享鎖的節點（讀操作）還是一個獨占鎖的節點（寫操作）。

上面說到的“酌情放棄這一次競爭”，ReentrantReadWriteLock中體現在boolean readerShouldBlock()這個方法里，這個方法有兩個模式：公平和非公平，我們來稍微看一點源碼
先看公平模式的實現：

final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}

當線程發現自己可以獲取讀鎖時（寫鎖未被占用），會調用這個方法，來判斷自己是否應該放棄此次獲取。
hasQueuedPredecessors()這個方法我們不去看源碼，因為它的意思很顯而易見（實際代碼也是）：是否存在排隊中的線程（Predecessor先驅者可以理解為先來的）。
如果有，那就放棄競爭去排隊。
在公平模式下，無論讀寫操作，只需要大家都遵守FIFO的秩序，就不會出現問題描述的情況

再來看看非公平模式下的實現代碼：

final boolean readerShouldBlock() {  
    return apparentlyFirstQueuedIsExclusive();
}

final boolean apparentlyFirstQueuedIsExclusive() {
    // Node表示同步隊列中的一個節點
    Node h, s;
    // head是當前隊列的頭節點的一個公共引用，它是一個沒有實際意義的節點，null or not只能標識隊列是否初始化過
    // next是當前節點的下一個節點的引用
    // isShared()方法表明這個節點是一個共享鎖（讀鎖）的節點還是獨占鎖（寫鎖）的節點
    return (h = head) != null &&
        (s = h.next)  != null &&
        !s.isShared()         &&
        s.thread != null;
}

總結一下語義：如果隊列不為空，且隊列最前面的節點是個獨占鎖的節點，則放棄競爭。也就是我們上面說的“根據隊列情況酌情放棄”。

一般而言的讀操作的線程對于state的操作可能只是+1然后-1，而如果發生重入，那就會是n次+1然后n次-1。
但是不管怎樣，每一個線程都應當有一份記錄自己持有共享鎖數量的信息，這樣釋放鎖的時候才能知道自己可不可以去-1。
這也許很簡單，我們可以在鎖里增加一個Map對象，用類似tid(k)-count(v)的數據結構來記錄每個線程的持有數量；也可以為每個線程創建一個ThreadLocal，讓它們自己拿著。

現在我們面前有兩條路比較直觀：將所有線程的小計數器維護在一個Map中，或是每個線程在ThreadLocal中維護自己的小計數器。
就這兩條途徑而言，應該是Map的這一條路比較高效，因為如果選擇ThreadLocal也許會頻繁進行其內部的ThreadLocalMap對象的創建和銷毀，這很消耗資源。

然而事實是，ReentranctReadWriteLock選擇的實現方式是后者，即使用ThreadLocal來實現，但是為什么選擇這種方式正是我十分好奇的地方，因為根據經驗，一定是利用Map統一管理小計數器的方式較為高效，且單個線程針對單個key的value進行+1或者-1的操作應該是滿足as-if-serial原則的，也不存在安全問題。
因此針對兩種不同的實現方式進行了一些測試：四線程并行情況下一千萬次加解鎖時間測試

Map統一管理實現

public static void main(String[] args) {
    long total = 0;
    for (int i = 0; i < 30; i++) {
        total += execute();
    }
    System.out.println(total / 30);
}

private static long execute() {

    var map = new HashMap();
    var readerPool = new ThreadPoolExecutor(
            4,
            4,
            5L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("readerPool"));

    var countDown = new CountDownLatch(10000000);
    for (int i = 0; i < 10000000; i++) {
        readerPool.execute(() -> {
            try {
                countDown.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
                return;
            }
            mapImplement(map);
        });
        countDown.countDown();
    }

    long startTime = System.currentTimeMillis();
    while (readerPool.getCompletedTaskCount() < 10000000) {
        LockSupport.parkNanos(100);
    }
    long total = System.currentTimeMillis() - startTime;

    System.out.println(readerPool.getCompletedTaskCount() + ", time: " + total);
    return total;
}

private static void mapImplement(HashMap map) {
    // lock
    var tid = Thread.currentThread().getId();
    Integer count;
    if ((count = map.get(tid)) != null) {
        map.put(tid, count + 1);
    } else {
        map.put(tid, 1);
    }

    // unlock
    int afterDecrement = -999;
    if ((count = map.get(tid)) == null ||
            (afterDecrement = (count - 1)) < 0) {
        System.out.println("error, count: " + count + ", afterDecrement: " + afterDecrement);
        return;
    }
    map.put(tid, afterDecrement);
}

三十次測試過程的平均執行時間為：2378毫秒，個人認為這個結果還是比較樂觀的。

ThreadLocal各自持有實現

// 小計數器實體
static final class HoldCounter {
    int count;
}

// threadLocal
static final class ThreadLocalHoldCounter extends ThreadLocal {
    @Override
    public HoldCounter initialValue() {
        return new HoldCounter();
    }
}

public static void main(String[] args) {
    long total = 0;
    for (int i = 0; i < 30; i++) {
        total += execute();
    }
    System.out.println(total / 30);
}

private static long execute() {
    
    var readHolds = new ThreadLocalHoldCounter();
    var readerPool = new ThreadPoolExecutor(
            4,
            4,
            5L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("readerPool"));

    var countDown = new CountDownLatch(10000000);
    for (int i = 0; i < 10000000; i++) {
        readerPool.execute(() -> {
            try {
                countDown.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
                return;
            }
            threadLocalImplement(readHolds);
        });
        countDown.countDown();
    }

    long startTime = System.currentTimeMillis();
    while (readerPool.getCompletedTaskCount() < 10000000) {
        LockSupport.parkNanos(100);
    }
    long total = System.currentTimeMillis() - startTime;

    System.out.println(readerPool.getCompletedTaskCount() + ", time: " + total);
    return total;
}

private static void threadLocalImplement(ThreadLocalHoldCounter readHolds) {
    // lock
    var hc = readHolds.get();
    ++hc.count;

    // unlock
    hc = readHolds.get();
    --hc.count;
    if (hc.count == 0) {
        readHolds.remove();
    }
}

三十次測試過程的平均執行時間為：3079毫秒

可以看到，使用Map集中管理小計數器的實現方式的執行效率要比ThreadLocal的實現方式快20%以上。

難道我作為一個Java萌新都能想到的性能差距，Doug Lea這樣的大神會想不到嗎？

當然不會

實際上，ReentranctReadWriteLock在針對小計數器的具體實現上，增加了類似兩層緩存的設計，大概如下：

// ThreadLocal對象本體
private transient ThreadLocalHoldCounter readHolds;

// 二級緩存：最近一次獲取鎖的線程所持有的小計數器對象的引用
private transient HoldCounter cachedHoldCounter;

// 一級緩存：首次獲取鎖的線程的線程對象引用以及它的計數
private transient Thread firstReader;
private transient int firstReaderHoldCount;

當線程嘗試獲取鎖時，會執行如下的流程：

判斷當前共享鎖總計數器是否為0（當前鎖處于空閑狀態） 或 firstReader == Thread.currentThread()
是： 則直接在firstReaderHoldCount上進行+1（以及執行firstReader = Thread.currentThread()）
否： 前往2

判斷cachedHoldCounter.tid == Thread.currentThread().getId()
是： 則直接在cachedHoldCounter.count上進行+1
否： 前往3

執行readHolds.get()進行獲取或初始化，然后再對小計數器進行操作

當線程釋放鎖時，執行流程也大致相似，都是先對兩級緩存進行嘗試，逼不得已再去對ThreadLocal進行操作。
由于讀操作的實際執行內容一般相當簡單（類似return a），所以在絕大多數情況下，線程的加解鎖行為都會命中一級緩存。

我嘗試在ReentranctReadWriteLock的加解鎖行為內埋了幾個計數點來測試兩級緩存的命中率，四線程并行1000萬次加解鎖操作，結果是：

一級緩存命中率大概為90~95%

二級緩存命中率大概為5~10%

ThreadLocal本體命中率大概為1~5%

而執行效率，進行1000萬次加解鎖，循環三十次得到的平均執行時間是：2027毫秒。
比上面提到的使用Map實現的方式更要快了15%左右。

雖說上面的小測試的編碼也好，測試環境也好，都不算特別嚴謹，但是還是能非常直觀地說明問題的吧。

如圖，ReentranctReadWriteLock中有五個內部類：

Sync
Sync繼承自AbstractQueuedSynchronizer，上文提到的volatile int state以及同步隊列的實際實現都是由AbstractQueuedSynchronizer這個抽象類提供的，它還提供了一些在鎖的性質不同時實現也會不同的可重寫方法，Sync需要做的事情就是將這些通用的方法和規則加以實現和擴充，形成自己想要實現的鎖。
Sync也是我們上文提到的，同時實現了讀寫兩種性質的鎖的根本。
另外，上文提到的關于分段使用state、利用公平性避免機會不均衡的問題、分級緩存共享鎖小計數器等特性，均在此類中實現，需要特別關注。

NonfairSync與FairSync
這兩個類都繼承自Sync，它們提供了由Sync定義的兩個用于進行公平性判斷的方法：boolean writerShouldBlock()與boolean readerShouldBlock()。實際使用ReentranctReadWriteLock時，我們會通過構造方法選擇需要構造公平還是非公平的鎖，相應的會通過這兩個子類構造實際的Sync類的對象，從而影響到加解鎖過程中的一些判斷。

ReadLock與WriteLock
這兩個類都會持有上面提到的Sync類的對象的引用，并向用戶（使用者）提供包裝好但實現不同的操作，比如：

讀鎖獲取

public void lock() {
    sync.acquireShared(1);
}

寫鎖獲取

public void lock() {
    sync.acquire(1);
}

更多的源碼就不再贅述了，搜索一下就會有非常多的文章解讀源碼并非作者懶得貼了。

juc這套并發框架的設計者和創始人Doug Lea，可以說是java開發者金字塔頂端的巨佬之一了，他所編寫的juc包的代碼，無論是代碼結構的合理性、各種設計模式的使用、代碼的優雅程度都令人嘆為觀止。看完源碼覺得整個人都升華了

筆者學習了源碼之后，覺得在面對這些充滿了考慮的設計細節時產生的思考，才是真正可以使人得到長遠的提升的東西。

因此整理出來，作為心得體會的記錄。如果可以對其他小伙伴帶來啟發，那就更好了。
最后，如果文章內有什么紕漏或是錯誤，還請務必指正，再次感謝。