1. 前言

在Java并發(fā)包中常用的鎖（如：ReentrantLock），基本上都是排他鎖，這些鎖在同一時(shí)刻只允許一個(gè)線程進(jìn)行訪問(wèn)，而讀寫(xiě)鎖在同一時(shí) 刻可以允許多個(gè)讀線程訪問(wèn)，但是在寫(xiě)線程訪問(wèn)時(shí)，所有的讀線程和其他寫(xiě)線程均被阻塞。讀寫(xiě)鎖維護(hù)了一對(duì)鎖，一個(gè)讀鎖和一個(gè)寫(xiě)鎖，通過(guò)分離讀鎖和寫(xiě)鎖，使得 并發(fā)性相比一般的排他鎖有了很大提升。

除了保證寫(xiě)操作對(duì)讀操作的可見(jiàn)性以及并發(fā)性的提升之外，讀寫(xiě)鎖能夠簡(jiǎn)化讀寫(xiě)交互場(chǎng)景的編程方式。假設(shè)在程序中定義一個(gè)共享的數(shù)據(jù)結(jié)構(gòu)用作緩存，它大部分時(shí)間提供讀服務(wù)（例如：查詢和搜索），而寫(xiě)操作占有的時(shí)間很少，但是寫(xiě)操作完成之后的更新需要對(duì)后續(xù)的讀服務(wù)可見(jiàn)。

在沒(méi)有讀寫(xiě)鎖支持的（Java 5 之前）時(shí)候，如果需要完成上述工作就要使用Java的等待通知機(jī)制，就是當(dāng)寫(xiě)操作開(kāi)始時(shí)，所有晚于寫(xiě)操作的讀操作均會(huì)進(jìn)入等待狀態(tài)，只有寫(xiě)操作完成并進(jìn)行 通知之后，所有等待的讀操作才能繼續(xù)執(zhí)行（寫(xiě)操作之間依靠synchronized關(guān)鍵字進(jìn)行同步），這樣做的目的是使讀操作都能讀取到正確的數(shù)據(jù)，而不 會(huì)出現(xiàn)臟讀。改用讀寫(xiě)鎖實(shí)現(xiàn)上述功能，只需要在讀操作時(shí)獲取讀鎖，而寫(xiě)操作時(shí)獲取寫(xiě)鎖即可，當(dāng)寫(xiě)鎖被獲取到時(shí)，后續(xù)（非當(dāng)前寫(xiě)操作線程）的讀寫(xiě)操作都會(huì)被 阻塞，寫(xiě)鎖釋放之后，所有操作繼續(xù)執(zhí)行，編程方式相對(duì)于使用等待通知機(jī)制的實(shí)現(xiàn)方式而言，變得簡(jiǎn)單明了。

一般情況下，讀寫(xiě)鎖的性能都會(huì)比排它鎖要好，因?yàn)榇蠖鄶?shù)場(chǎng)景讀是多于寫(xiě)的。在讀多于寫(xiě)的情況下，讀寫(xiě)鎖能夠提供比排它鎖更好的并發(fā)性和吞吐量。Java并發(fā)包提供讀寫(xiě)鎖的實(shí)現(xiàn)是ReentrantReadWriteLock，它提供的特性如表1所示。

表1. ReentrantReadWriteLock的特性

2. 讀寫(xiě)鎖的接口與示例

ReadWriteLock僅定義了獲取讀鎖和寫(xiě)鎖的兩個(gè)方法，即readLock()和writeLock()方法，而其實(shí)現(xiàn)— ReentrantReadWriteLock，除了接口方法之外，還提供了一些便于外界監(jiān)控其內(nèi)部工作狀態(tài)的方法，這些方法以及描述如表2所示。

表2. ReentrantReadWriteLock展示內(nèi)部工作狀態(tài)的方法

接下來(lái)通過(guò)一個(gè)緩存示例說(shuō)明讀寫(xiě)鎖的使用方式，示例代碼如代碼清單1所示。

代碼清單1. Cache.java

public class Cache { 
  static Map<String, Object> map = new HashMap<String, Object>(); 
  static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); 
  static Lock r = rwl.readLock(); 
  static Lock w = rwl.writeLock(); 
  // 獲取一個(gè)key對(duì)應(yīng)的value 
  public static final Object get(String key) { 
    r.lock(); 
    try { 
      return map.get(key); 
    } finally { 
      r.unlock(); 
    } 
  } 
  // 設(shè)置key對(duì)應(yīng)的value，并返回舊有的value 
  public static final Object put(String key, Object value) { 
    w.lock(); 
    try { 
      return map.put(key, value); 
    } finally { 
      w.unlock(); 
    } 
  } 
  // 清空所有的內(nèi)容 
  public static final void clear() { 
    w.lock(); 
    try { 
      map.clear(); 
    } finally { 
      w.unlock(); 
    } 
  } 
} 

上述示例中，Cache組合了一個(gè)非線程安全的HashMap作為緩存的實(shí)現(xiàn)，同時(shí)使用讀寫(xiě)鎖的讀鎖和寫(xiě)鎖來(lái)保證Cache是線程安全的。在讀操作 get(String key)方法中，需要獲取讀鎖，這使得并發(fā)訪問(wèn)該方法時(shí)不會(huì)被阻塞。寫(xiě)操作put(String key, Object value)和clear()方法，在更新HashMap時(shí)必須提前獲取寫(xiě)鎖，當(dāng)寫(xiě)鎖被獲取后，其他線程對(duì)于讀鎖和寫(xiě)鎖的獲取均被阻塞，而只有寫(xiě)鎖被釋放 之后，其他讀寫(xiě)操作才能繼續(xù)。Cache使用讀寫(xiě)鎖提升讀操作并發(fā)性，也保證每次寫(xiě)操作對(duì)所有的讀寫(xiě)操作的可見(jiàn)性，同時(shí)簡(jiǎn)化了編程方式。

3. 讀寫(xiě)鎖的實(shí)現(xiàn)分析

接下來(lái)將分析ReentrantReadWriteLock的實(shí)現(xiàn)，主要包括：讀寫(xiě)狀態(tài)的設(shè)計(jì)、寫(xiě)鎖的獲取與釋放、讀鎖的獲取與釋放以及鎖降級(jí)（以下沒(méi)有特別說(shuō)明讀寫(xiě)鎖均可認(rèn)為是ReentrantReadWriteLock）。

3.1 讀寫(xiě)狀態(tài)的設(shè)計(jì)

讀寫(xiě)鎖同樣依賴自定義同步器來(lái)實(shí)現(xiàn)同步功能，而讀寫(xiě)狀態(tài)就是其同步器的同步狀態(tài)?；叵隦eentrantLock中自定義同步器的實(shí)現(xiàn)，同步狀態(tài) 表示鎖被一個(gè)線程重復(fù)獲取的次數(shù)，而讀寫(xiě)鎖的自定義同步器需要在同步狀態(tài)（一個(gè)整型變量）上維護(hù)多個(gè)讀線程和一個(gè)寫(xiě)線程的狀態(tài)，使得該狀態(tài)的設(shè)計(jì)成為讀寫(xiě) 鎖實(shí)現(xiàn)的關(guān)鍵。

如果在一個(gè)整型變量上維護(hù)多種狀態(tài)，就一定需要“按位切割使用”這個(gè)變量，讀寫(xiě)鎖是將變量切分成了兩個(gè)部分，高16位表示讀，低16位表示寫(xiě)，劃分方式如圖1所示。

圖1. 讀寫(xiě)鎖狀態(tài)的劃分方式

如圖1所示，當(dāng)前同步狀態(tài)表示一個(gè)線程已經(jīng)獲取了寫(xiě)鎖，且重進(jìn)入了兩次，同時(shí)也連續(xù)獲取了兩次讀鎖。讀寫(xiě)鎖是如何迅速的確定讀和寫(xiě)各自的狀態(tài)呢？ 答案是通過(guò)位運(yùn)算。假設(shè)當(dāng)前同步狀態(tài)值為S，寫(xiě)狀態(tài)等于 S & 0x0000FFFF（將高16位全部抹去），讀狀態(tài)等于 S >>> 16（無(wú)符號(hào)補(bǔ)0右移16位）。當(dāng)寫(xiě)狀態(tài)增加1時(shí)，等于S + 1，當(dāng)讀狀態(tài)增加1時(shí)，等于S + (1 << 16)，也就是S + 0×00010000。

根據(jù)狀態(tài)的劃分能得出一個(gè)推論：S不等于0時(shí)，當(dāng)寫(xiě)狀態(tài)(S & 0x0000FFFF)等于0時(shí)，則讀狀態(tài)(S >>> 16)大于0，即讀鎖已被獲取。

3.2 寫(xiě)鎖的獲取與釋放

寫(xiě)鎖是一個(gè)支持重進(jìn)入的排它鎖。如果當(dāng)前線程已經(jīng)獲取了寫(xiě)鎖，則增加寫(xiě)狀態(tài)。如果當(dāng)前線程在獲取寫(xiě)鎖時(shí)，讀鎖已經(jīng)被獲?。ㄗx狀態(tài)不為0）或者該線程不是已經(jīng)獲取寫(xiě)鎖的線程，則當(dāng)前線程進(jìn)入等待狀態(tài)，獲取寫(xiě)鎖的代碼如代碼清單2所示。

代碼清單2. ReentrantReadWriteLock的tryAcquire方法

protected final boolean tryAcquire(int acquires) { 
  Thread current = Thread.currentThread(); 
  int c = getState(); 
  int w = exclusiveCount(c); 
  if (c != 0) { 
    // 存在讀鎖或者當(dāng)前獲取線程不是已經(jīng)獲取寫(xiě)鎖的線程 
    if (w == 0 || current != getExclusiveOwnerThread()) 
      return false; 
    if (w + exclusiveCount(acquires) > MAX_COUNT) 
      throw new Error("Maximum lock count exceeded"); 
    setState(c + acquires); 
    return true; 
  } 
  if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) { 
    return false; 
  } 
  setExclusiveOwnerThread(current); 
  return true; 
} 

該方法除了重入條件（當(dāng)前線程為獲取了寫(xiě)鎖的線程）之外，增加了一個(gè)讀鎖是否存在的判斷。如果存在讀鎖，則寫(xiě)鎖不能被獲取，原因在于：讀寫(xiě)鎖要確保 寫(xiě)鎖的操作對(duì)讀鎖可見(jiàn)，如果允許讀鎖在已被獲取的情況下對(duì)寫(xiě)鎖的獲取，那么正在運(yùn)行的其他讀線程就無(wú)法感知到當(dāng)前寫(xiě)線程的操作。因此只有等待其他讀線程都 釋放了讀鎖，寫(xiě)鎖才能被當(dāng)前線程所獲取，而寫(xiě)鎖一旦被獲取，則其他讀寫(xiě)線程的后續(xù)訪問(wèn)均被阻塞。

寫(xiě)鎖的釋放與ReentrantLock的釋放過(guò)程基本類似，每次釋放均減少寫(xiě)狀態(tài)，當(dāng)寫(xiě)狀態(tài)為0時(shí)表示寫(xiě)鎖已被釋放，從而等待的讀寫(xiě)線程能夠繼續(xù)訪問(wèn)讀寫(xiě)鎖，同時(shí)前次寫(xiě)線程的修改對(duì)后續(xù)讀寫(xiě)線程可見(jiàn)。

3.3 讀鎖的獲取與釋放

讀鎖是一個(gè)支持重進(jìn)入的共享鎖，它能夠被多個(gè)線程同時(shí)獲取，在沒(méi)有其他寫(xiě)線程訪問(wèn)（或者寫(xiě)狀態(tài)為0）時(shí)，讀鎖總會(huì)成功的被獲取，而所做的也只是 （線程安全的）增加讀狀態(tài)。如果當(dāng)前線程已經(jīng)獲取了讀鎖，則增加讀狀態(tài)。如果當(dāng)前線程在獲取讀鎖時(shí)，寫(xiě)鎖已被其他線程獲取，則進(jìn)入等待狀態(tài)。獲取讀鎖的實(shí) 現(xiàn)從Java 5到Java 6變得復(fù)雜許多，主要原因是新增了一些功能，比如：getReadHoldCount()方法，返回當(dāng)前線程獲取讀鎖的次數(shù)。讀狀態(tài)是所有線程獲取讀鎖次 數(shù)的總和，而每個(gè)線程各自獲取讀鎖的次數(shù)只能選擇保存在ThreadLocal中，由線程自身維護(hù)，這使獲取讀鎖的實(shí)現(xiàn)變得復(fù)雜。因此，這里將獲取讀鎖的 代碼做了刪減，保留必要的部分，代碼如代碼清單3所示。

代碼清單3. ReentrantReadWriteLock的tryAcquireShared方法

protected final int tryAcquireShared(int unused) { 
  for (;;) { 
    int c = getState(); 
    int nextc = c + (1 << 16); 
    if (nextc < c) 
      throw new Error("Maximum lock count exceeded"); 
    if (exclusiveCount(c) != 0 && owner != Thread.currentThread()) 
      return -1; 
    if (compareAndSetState(c, nextc)) 
      return 1; 
  } 
} 

在tryAcquireShared(int unused)方法中，如果其他線程已經(jīng)獲取了寫(xiě)鎖，則當(dāng)前線程獲取讀鎖失敗，進(jìn)入等待狀態(tài)。如果當(dāng)前線程獲取了寫(xiě)鎖或者寫(xiě)鎖未被獲取，則當(dāng)前線程(線程安全，依靠CAS保證)增加讀狀態(tài)，成功獲取讀鎖。

讀鎖的每次釋放均（線程安全的，可能有多個(gè)讀線程同時(shí)釋放讀鎖）減少讀狀態(tài)，減少的值是(1 << 16)。

3.4 鎖降級(jí)

鎖降級(jí)指的是寫(xiě)鎖降級(jí)成為讀鎖。如果當(dāng)前線程擁有寫(xiě)鎖，然后將其釋放，***再獲取讀鎖，這種分段完成的過(guò)程不能稱之為鎖降級(jí)。鎖降級(jí)是指把持?。ó?dāng)前擁有的）寫(xiě)鎖，再獲取到讀鎖，隨后釋放（先前擁有的）寫(xiě)鎖的過(guò)程。

接下來(lái)看一個(gè)鎖降級(jí)的示例：因?yàn)閿?shù)據(jù)不常變化，所以多個(gè)線程可以并發(fā)的進(jìn)行數(shù)據(jù)處理，當(dāng)數(shù)據(jù)變更后，當(dāng)前線程如果感知到數(shù)據(jù)變化，則進(jìn)行數(shù)據(jù)的準(zhǔn)備工作，同時(shí)其他處理線程被阻塞，直到當(dāng)前線程完成數(shù)據(jù)的準(zhǔn)備工作，示例代碼如代碼清單4所示。

代碼清單4. processData方法

public void processData() { 
  readLock.lock(); 
  if (!update) { 
    // 必須先釋放讀鎖 
    readLock.unlock(); 
    // 鎖降級(jí)從寫(xiě)鎖獲取到開(kāi)始 
    writeLock.lock(); 
    try { 
      if (!update) { 
        // 準(zhǔn)備數(shù)據(jù)的流程（略） 
        update = true; 
      } 
      readLock.lock(); 
    } finally { 
      writeLock.unlock(); 
    } 
    // 鎖降級(jí)完成，寫(xiě)鎖降級(jí)為讀鎖 
  } 
  try { 
    // 使用數(shù)據(jù)的流程（略） 
  } finally { 
    readLock.unlock(); 
  } 
} 

上述示例中，當(dāng)數(shù)據(jù)發(fā)生變更后，update變量（布爾類型且Volatile修飾）被設(shè)置為false，此時(shí)所有訪問(wèn)processData() 方法的線程都能夠感知到變化，但只有一個(gè)線程能夠獲取到寫(xiě)鎖，而其他線程會(huì)被阻塞在讀鎖和寫(xiě)鎖的lock()方法上。當(dāng)前程獲取寫(xiě)鎖完成數(shù)據(jù)準(zhǔn)備之后，再 獲取讀鎖，隨后釋放寫(xiě)鎖，完成鎖降級(jí)。

鎖降級(jí)中讀鎖的獲取是否必要呢？答案是必要的。主要原因是保證數(shù)據(jù)的可見(jiàn)性，如果當(dāng)前線程不獲取讀鎖而是直接釋放寫(xiě)鎖，假設(shè)此刻另一個(gè)線程（記作 線程T）獲取了寫(xiě)鎖并修改了數(shù)據(jù)，則當(dāng)前線程無(wú)法感知線程T的數(shù)據(jù)更新。如果當(dāng)前線程獲取讀鎖，即遵循鎖降級(jí)的步驟，則線程T將會(huì)被阻塞，直到當(dāng)前線程使 用數(shù)據(jù)并釋放讀鎖之后，線程T才能獲取寫(xiě)鎖進(jìn)行數(shù)據(jù)更新。

RentrantReadWriteLock不支持鎖升級(jí)（把持讀鎖、獲取寫(xiě)鎖，***釋放讀鎖的過(guò)程）。原因也是保證數(shù)據(jù)可見(jiàn)性，如果讀鎖已被多個(gè)線程獲取，其中任意線程成功獲取了寫(xiě)鎖并更新了數(shù)據(jù)，則其更新對(duì)其他獲取到讀鎖的線程不可見(jiàn)。