[[396048]]

 引言

前文我們有介紹《看了CopyOnWriteArrayList后自己實現了一個CopyOnWriteHashMap》 關于CopyOnWrite容器的，但是它也有一些缺點：

• 內存占用問題：因為CopyOnWrite的寫時復制機制每次進行寫操作的時候都會有兩個數組對象的內存，如果這個數組對象占用的內存較大的話，如果頻繁的進行寫入就會造成頻繁的Yong GC和Full GC
• 數據一致性問題：CopyOnWrite容器只能保證數據的最終一致性，不能保證數據的實時一致性。讀操作的線程可能不會立即讀取到新修改的數據，因為修改操作發生在副本上。但最終修改操作會完成并更新容器所以這是最終一致性。當時有說到解決這兩個缺點我們可以使用Collections.synchronizedList()來替代，找個無非就是對list的增刪改查方法都加了synchronized實現。我們知道synchronized其實是一個獨占鎖 (排他鎖)，如果不知道什么是獨占鎖的可以看看這個文章《史上最全 Java 中各種鎖的介紹》 里面基本上把java里面的鎖都介紹完了。但是這樣的話就會存在一個性能問題，如果對于讀多寫少的場景，每次讀也要去獲取鎖，讀完了之后再釋放鎖，這樣就造成了每個讀的請求都要進行獲取鎖，但是讀的話并不會引起數據不安全，這樣就會造成一個性能瓶頸。為了解決這個問題，就又出現了一種新的鎖，讀寫鎖(ReadWriteLock)。

什么是讀寫鎖

根據名字我們也可以猜個大概，就是有兩把鎖，分別是讀鎖和寫鎖。讀鎖在同一時刻可以允許多個讀線程獲取，但是在寫線程訪問的時候，所有的讀線程和其他寫線程都會被阻塞。寫鎖同一時刻只能有一個寫線程獲取成功，其他都會被阻塞。讀寫鎖實際維護了兩把鎖，一個讀鎖和一個寫鎖，通過讀鎖和寫鎖進行區分，在讀多寫少的情況下并發性比獨占鎖有了很大的提升。在java里面對讀寫鎖的實現就是ReentrantReadWriteLock，它有以下特性：

• 公平性選擇：支持非公平性(默認)和公平的鎖獲取方式，吞吐量還是非公平優于公平;
• 重入性：支持重入，讀鎖獲取后能再次獲取，寫鎖獲取之后能夠再次獲取寫鎖，同時也能夠獲取讀鎖;
• 鎖降級：遵循獲取寫鎖，獲取讀鎖再釋放寫鎖的次序，寫鎖能夠降級成為讀鎖

ReentrantReadWriteLock 的使用

我們先從官網來個事例https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/locks/ReentrantReadWriteLock.html，看看它是如何使用的

class RWDictionary { 
  private final Map<String, Data> m = new TreeMap<String, Data>(); 
  private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); 
  private final Lock r = rwl.readLock(); 
  private final Lock w = rwl.writeLock(); 
 
  public Data get(String key) { 
    r.lock(); 
    try { return m.get(key); } 
    finally { r.unlock(); } 
  } 
  public String[] allKeys() { 
    r.lock(); 
    try { return m.keySet().toArray(); } 
    finally { r.unlock(); } 
  } 
  public Data put(String key, Data value) { 
    w.lock(); 
    try { return m.put(key, value); } 
    finally { w.unlock(); } 
  } 
  public void clear() { 
    w.lock(); 
    try { m.clear(); } 
    finally { w.unlock(); } 
  } 
} 

這個使用起來還是非常簡單明了的，跟ReentrantLock的用法基本一致，寫的時候獲取寫鎖，寫完了釋放寫鎖，讀的時候獲取讀鎖，讀完了就釋放讀寫。

讀寫鎖的實現分析

我們知道ReentrantLock是通過state來控制鎖的狀態，以及前面所介紹的《Java高并發編程基礎三大利器之Semaphore》《Java高并發編程基礎三大利器之CountDownLatch》《Java高并發編程基礎三大利器之CyclicBarrier》 都是通過state來進行實現的那ReentrantReadWriteLock毋庸置疑肯定也是通過AQS的state來實現的，不過state是一個int值它是如何來讀鎖和寫鎖的。

讀寫鎖狀態的實現分析

如果我們有看過線程池的源碼，我們知道線程池的狀態和線程數是通過一個int類型原子變量(高3位保存運行狀態，低29位保存線程數)來控制的。同樣的ReentrantReadWriteLock也是通過一個state的高16位和低16位來分別控制讀的狀態和寫狀態。

下面我們就來看看它是如何通過一個字段來實現讀寫分離的,

static final int SHARED_SHIFT   = 16; 
 static final int SHARED_UNIT    = (1 << SHARED_SHIFT); 
 static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1; 
 static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; 
 
 /** Returns the number of shared holds represented in count  */ 
 static int sharedCount(int c)    { return c >>> SHARED_SHIFT; } 
 /** Returns the number of exclusive holds represented in count  */ 
 static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; } 

• sharedCount : 讀鎖數量 是將同步狀態(int c)無符號右移16位，即取同步狀態的高16位。
• exclusiveCount：寫鎖數量 我們要看下EXCLUSIVE_MASK 這個靜態變量：它是1進行左移16位然后減1也就是0X0000FFFF即 (1 << SHARED_SHIFT) - 1= 0X0000FFFF 所以exclusiveCount 就是相當于 c&0X0000FFFF 所以也就是低16位用來表示寫鎖的獲取次數。

源碼分析

基于jdk1.8 既然ReentrantReadWriteLock也是基于AQS來實現的，那么它肯定是重寫了AQS的獲取鎖的方法，那我們就直接去ReentrantReadWriteLock這個類里面看看lock的地方我們先看看獲取讀鎖的地方

protected final boolean tryAcquire(int acquires) { 
          /* 
           * Walkthrough: 
           * 1. If read count nonzero or write count nonzero 
           *    and owner is a different thread, fail. 
           * 2. If count would saturate, fail. (This can only 
           *    happen if count is already nonzero.) 
           * 3. Otherwise, this thread is eligible for lock if 
           *    it is either a reentrant acquire or 
           *    queue policy allows it. If so, update state 
           *    and set owner. 
           */ 
          Thread current = Thread.currentThread(); 
          // 獲取寫鎖當前的同步狀態 
          int c = getState(); 
          // 寫鎖次數 
          int w = exclusiveCount(c); 
          if (c != 0) { 
              // (Note: if c != 0 and w == 0 then shared count != 0) 
              // 當前狀態不為0，但是寫鎖為0 就說明讀鎖不為0 
             // 當讀鎖已被讀線程獲取或者當前線程不是已經獲取寫鎖的線程的話獲取寫鎖失敗 
              if (w == 0 || current != getExclusiveOwnerThread()) 
                  return false; 
              if (w + exclusiveCount(acquires) > MAX_COUNT) 
                  throw new Error("Maximum lock count exceeded"); 
              // Reentrant acquire 獲取到寫鎖 
              setState(c + acquires); 
              return true; 
          } 
             //writerShouldBlock 公平鎖和非公平鎖的判斷 
          if (writerShouldBlock() || 
              !compareAndSetState(c, c + acquires)) 
              return false; 
          setExclusiveOwnerThread(current); 
          return true; 
      } 

寫鎖完了，接下來肯定就是讀鎖了由于讀鎖是共享鎖，所以也應該重寫了tryAcquireShared 這個就不貼代碼了，和讀鎖差不多這個就不做分析了。其實把AQS弄明白了再來看這些基于AQS來實現的玩意還是比較容易的。

讀寫鎖的升級與降級

前面我們有提到讀寫鎖是可以降級的，但是沒有說是否可以升級。我們先看看什么是鎖降級和鎖升級

• 鎖降級：從寫鎖變成讀鎖;它的過程是先持有寫鎖，在獲取讀鎖，再釋放寫鎖。如果是持有寫鎖，釋放寫鎖，再獲取讀鎖這種情況不是鎖降級。



• 為什么要鎖降級?

主要是為了保證數據的可見性，如果當前線程不獲取讀鎖而是直接釋放寫鎖， 假設此刻另一個線程(記作線程T)獲取了寫鎖并修改了數據，那么當前線程無法感知線程T的數據更新。如果當前線程獲取讀鎖，即遵循鎖降級的步驟，則線程T將會被阻塞，直到當前線程使用數據并釋放讀鎖之后，線程T才能獲取寫鎖進行數據更新。來源于《Java 并發編程的藝術》”

• 鎖升級：從讀鎖變成寫鎖。先持有讀鎖，再去獲取寫鎖(這是不會成功的)因為獲取寫鎖是獨占鎖，如果有讀鎖被占用了，寫鎖就會放入隊列中等待，直至讀鎖全部被釋放之后才有可能獲取到寫鎖。

思考題

• 本篇文章主要介紹了單機情況的讀寫鎖，如果要實現一個分布式的讀寫鎖該如何實現?
• ReentrantReadWriteLock的饑餓問題如何解決?(ReentrantReadWriteLock實現了讀寫分離，想要獲取讀鎖就必須確保當前沒有其他任何讀寫鎖了，但是一旦讀操作比較多的時候，想要獲取寫鎖就變得比較困難了，因為當前有可能會一直存在讀鎖。而無法獲得寫鎖。)

結束

由于自己才疏學淺，難免會有紕漏，假如你發現了錯誤的地方，還望留言給我指出來,我會對其加以修正。

本文轉載自微信公眾號「java金融」，可以通過以下二維碼關注。轉載本文請聯系java金融公眾號。