Java 讀寫鎖實現(xiàn)原理淺析

更新時間：2018年08月16日 09:17:07 作者：編輯部的故事

這篇文章主要介紹了Java 讀寫鎖實現(xiàn)原理淺析,非常不錯，具有一定的參考借鑒價值，需要的朋友可以參考下

最近做的一個小項目中有這樣的需求：整個項目有一份config.json保存著項目的一些配置，是存儲在本地文件的一個資源，并且應(yīng)用中存在讀寫（讀>>寫）更新問題。既然讀寫并發(fā)操作，那么就涉及到操作互斥，這里自然想到了讀寫鎖，本文對讀寫鎖方面的知識做個梳理。

為什么需要讀寫鎖？

與傳統(tǒng)鎖不同的是讀寫鎖的規(guī)則是可以共享讀，但只能一個寫，總結(jié)起來為：讀讀不互斥，讀寫互斥，寫寫互斥，而一般的獨占鎖是：讀讀互斥，讀寫互斥，寫寫互斥，而場景中往往讀遠遠大于寫，讀寫鎖就是為了這種優(yōu)化而創(chuàng)建出來的一種機制。

注意是讀遠遠大于寫，一般情況下獨占鎖的效率低來源于高并發(fā)下對臨界區(qū)的激烈競爭導(dǎo)致線程上下文切換。因此當(dāng)并發(fā)不是很高的情況下，讀寫鎖由于需要額外維護讀鎖的狀態(tài)，可能還不如獨占鎖的效率高。因此需要根據(jù)實際情況選擇使用。

一個簡單的讀寫鎖實現(xiàn)

根據(jù)上面理論可以利用兩個int變量來簡單實現(xiàn)一個讀寫鎖，實現(xiàn)雖然爛，但是原理都是差不多的，值得閱讀下。

public class ReadWriteLock {
 /**
  * 讀鎖持有個數(shù)
  */
 private int readCount = 0;
 /**
  * 寫鎖持有個數(shù)
  */
 private int writeCount = 0;
 /**
  * 獲取讀鎖,讀鎖在寫鎖不存在的時候才能獲取
  */
 public synchronized void lockRead() throws InterruptedException {
  // 寫鎖存在,需要wait
  while (writeCount > 0) {
   wait();
  }
  readCount++;
 }
 /**
  * 釋放讀鎖
  */
 public synchronized void unlockRead() {
  readCount--;
  notifyAll();
 }
 /**
  * 獲取寫鎖,當(dāng)讀鎖存在時需要wait.
  */
 public synchronized void lockWrite() throws InterruptedException {
  // 先判斷是否有寫請求
  while (writeCount > 0) {
   wait();
  }
  // 此時已經(jīng)不存在獲取寫鎖的線程了,因此占坑,防止寫鎖饑餓
  writeCount++;
  // 讀鎖為0時獲取寫鎖
  while (readCount > 0) {
   wait();
  }
 }
 /**
  * 釋放讀鎖
  */
 public synchronized void unlockWrite() {
  writeCount--;
  notifyAll();
 }
}

ReadWriteLock的實現(xiàn)原理

在Java中ReadWriteLock的主要實現(xiàn)為ReentrantReadWriteLock，其提供了以下特性：

公平性選擇：支持公平與非公平（默認）的鎖獲取方式，吞吐量非公平優(yōu)先于公平。
可重入：讀線程獲取讀鎖之后可以再次獲取讀鎖，寫線程獲取寫鎖之后可以再次獲取寫鎖
可降級：寫線程獲取寫鎖之后，其還可以再次獲取讀鎖，然后釋放掉寫鎖，那么此時該線程是讀鎖狀態(tài)，也就是降級操作。

ReentrantReadWriteLock的結(jié)構(gòu)

ReentrantReadWriteLock的核心是由一個基于AQS的同步器Sync構(gòu)成，然后由其擴展出ReadLock（共享鎖），WriteLock（排它鎖）所組成。

并且從ReentrantReadWriteLock的構(gòu)造函數(shù)中可以發(fā)現(xiàn)ReadLock與WriteLock使用的是同一個Sync，具體怎么實現(xiàn)同一個隊列既可以為共享鎖，又可以表示排他鎖下文會具體分析。

清單一：ReentrantReadWriteLock構(gòu)造函數(shù)

public ReentrantReadWriteLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
    readerLock = new ReadLock(this);
    writerLock = new WriteLock(this);
  }

Sync的實現(xiàn)

sync是讀寫鎖實現(xiàn)的核心，sync是基于AQS實現(xiàn)的，在AQS中核心是state字段和雙端隊列，那么一個一個問題來分析。

Sync如何同時表示讀鎖與寫鎖？

清單2：讀寫鎖狀態(tài)獲取

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; }

從代碼中獲取讀寫狀態(tài)可以看出其是把state（int32位）字段分成高16位與低16位，其中高16位表示讀鎖個數(shù)，低16位表示寫鎖個數(shù)，如下圖所示（圖來自Java并發(fā)編程藝術(shù)）。

該圖表示當(dāng)前一個線程獲取到了寫鎖，并且重入了兩次，因此低16位是3，并且該線程又獲取了讀鎖，并且重入了一次，所以高16位是2，當(dāng)寫鎖被獲取時如果讀鎖不為0那么讀鎖一定是獲取寫鎖的這個線程。

讀鎖的獲取

讀鎖的獲取主要實現(xiàn)是AQS中的acquireShared方法，其調(diào)用過程如下代碼。

清單3：讀鎖獲取入口

// ReadLock
public void lock() {
  sync.acquireShared(1);
}
// AQS
public final void acquireShared(int arg) {
  if (tryAcquireShared(arg) < 0)
    doAcquireShared(arg);
}

其中doAcquireShared(arg)方法是獲取失敗之后AQS中入隊操作，等待被喚醒后重新獲取，那么關(guān)鍵點就是tryAcquireShared(arg)方法，方法有點長，因此先總結(jié)出獲取讀鎖所經(jīng)歷的步驟，獲取的第一部分步驟如下：

操作1：讀寫需要互斥，因此當(dāng)存在寫鎖并且持有寫鎖的線程不是該線程時獲取失敗。
操作2：是否存在等待寫鎖的線程，存在的話則獲取讀鎖需要等待，避免寫鎖饑餓。(寫鎖優(yōu)先級是比較高的)
操作3：CAS獲取讀鎖，實際上是state字段的高16位自增。
操作4：獲取成功后再ThreadLocal中記錄當(dāng)前線程獲取讀鎖的次數(shù)。

清單4：讀鎖獲取的第一部分

protected final int tryAcquireShared(int unused) {
     Thread current = Thread.currentThread();
     int c = getState();
     // 操作1：存在寫鎖，并且寫鎖不是當(dāng)前線程則直接去排隊
     if (exclusiveCount(c) != 0 &&
       getExclusiveOwnerThread() != current)
       return -1;

     int r = sharedCount(c);
     // 操作2：讀鎖是否該阻塞，對于非公平模式下寫鎖獲取優(yōu)先級會高，如果存在要獲取寫鎖的線程則讀鎖需要讓步，公平模式下則先來先到
     if (!readerShouldBlock() && 
       // 讀鎖使用高16位，因此存在獲取上限為2^16-1
       r < MAX_COUNT &&
       // 操作3：CAS修改讀鎖狀態(tài)，實際上是讀鎖狀態(tài)+1
       compareAndSetState(c, c + SHARED_UNIT)) {
       // 操作4：執(zhí)行到這里說明讀鎖已經(jīng)獲取成功，因此需要記錄線程狀態(tài)。
       if (r == 0) {
         firstReader = current; // firstReader是把讀鎖狀態(tài)從0變成1的那個線程
         firstReaderHoldCount = 1;
       } else if (firstReader == current) { 
         firstReaderHoldCount++;
       } else {
         // 這些代碼實際上是從ThreadLocal中獲取當(dāng)前線程重入讀鎖的次數(shù)，然后自增下。
         HoldCounter rh = cachedHoldCounter; // cachedHoldCounter是上一個獲取鎖成功的線程
         if (rh == null || rh.tid != getThreadId(current))
           cachedHoldCounter = rh = readHolds.get();
         else if (rh.count == 0)
           readHolds.set(rh);
         rh.count++;
       }
       return 1;
     }
     // 當(dāng)操作2，操作3失敗時執(zhí)行該邏輯
     return fullTryAcquireShared(current);
   }

當(dāng)操作2，操作3失敗時會執(zhí)行fullTryAcquireShared(current)，為什么會這樣寫呢？個人認為是一種補償操作，操作2與操作3失敗并不代表當(dāng)前線程沒有讀鎖的資格，并且這里的讀鎖是共享鎖，有資格就應(yīng)該被獲取成功，因此給予補償獲取讀鎖的操作。在fullTryAcquireShared(current)中是一個循環(huán)獲取讀鎖的過程，大致步驟如下：

操作5：等同于操作2，存在寫鎖，且寫鎖線程并非當(dāng)前線程則直接返回失敗
操作6：當(dāng)前線程是重入讀鎖，這里只會偏向第一個獲取讀鎖的線程以及最后一個獲取讀鎖的線程，其他都需要去AQS中排隊。
操作7：CAS改變讀鎖狀態(tài)
操作8：同操作4，獲取成功后再ThreadLocal中記錄當(dāng)前線程獲取讀鎖的次數(shù)。

清單5：讀鎖獲取的第二部分

final int fullTryAcquireShared(Thread current) {
      HoldCounter rh = null;
      // 最外層嵌套循環(huán)
      for (;;) {
        int c = getState();
        // 操作5：存在寫鎖，且寫鎖并非當(dāng)前線程則直接返回失敗
        if (exclusiveCount(c) != 0) {
          if (getExclusiveOwnerThread() != current)
            return -1;
          // else we hold the exclusive lock; blocking here
          // would cause deadlock.
        // 操作6：如果當(dāng)前線程是重入讀鎖則放行
        } else if (readerShouldBlock()) {
          // Make sure we're not acquiring read lock reentrantly
          // 當(dāng)前是firstReader，則直接放行,說明是已獲取的線程重入讀鎖
          if (firstReader == current) {
            // assert firstReaderHoldCount > 0;
          } else {
            // 執(zhí)行到這里說明是其他線程，如果是cachedHoldCounter（其count不為0）也就是上一個獲取鎖的線程則可以重入，否則進入AQS中排隊
            // **這里也是對寫鎖的讓步**，如果隊列中頭結(jié)點為寫鎖，那么當(dāng)前獲取讀鎖的線程要進入隊列中排隊
            if (rh == null) {
              rh = cachedHoldCounter;
              if (rh == null || rh.tid != getThreadId(current)) {
                rh = readHolds.get();
                if (rh.count == 0)
                  readHolds.remove();
              }
            }
            // 說明是上述剛初始化的rh，所以直接去AQS中排隊
            if (rh.count == 0)
              return -1;
          }
        }
        if (sharedCount(c) == MAX_COUNT)
          throw new Error("Maximum lock count exceeded");
        // 操作7：修改讀鎖狀態(tài)，實際上讀鎖自增操作
        if (compareAndSetState(c, c + SHARED_UNIT)) {
          // 操作8：對ThreadLocal中維護的獲取鎖次數(shù)進行更新。
          if (sharedCount(c) == 0) {
            firstReader = current;
            firstReaderHoldCount = 1;
          } else if (firstReader == current) {
            firstReaderHoldCount++;
          } else {
            if (rh == null)
              rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
              rh = readHolds.get();
            else if (rh.count == 0)
              readHolds.set(rh);
            rh.count++;
            cachedHoldCounter = rh; // cache for release
          }
          return 1;
        }
      }
    }

讀鎖的釋放

清單6：讀鎖釋放入口

// ReadLock
public void unlock() {
  sync.releaseShared(1);
}
// Sync
public final boolean releaseShared(int arg) {
  if (tryReleaseShared(arg)) {
    doReleaseShared(); // 這里實際上是釋放讀鎖后喚醒寫鎖的線程操作
    return true;
  }
  return false;
}

讀鎖的釋放主要是tryReleaseShared(arg)函數(shù)，因此拆解其步驟如下：

操作1：清理ThreadLocal中保存的獲取鎖數(shù)量信息
操作2：CAS修改讀鎖個數(shù)，實際上是自減一

清單7：讀鎖的釋放流程

protected final boolean tryReleaseShared(int unused) {
     Thread current = Thread.currentThread();
     // 操作1：清理ThreadLocal對應(yīng)的信息
     if (firstReader == current) {;
       if (firstReaderHoldCount == 1)
         firstReader = null;
       else
         firstReaderHoldCount--;
     } else {
       HoldCounter rh = cachedHoldCounter;
       if (rh == null || rh.tid != getThreadId(current))
         rh = readHolds.get();
       int count = rh.count;
       // 已釋放完的讀鎖的線程清空操作
       if (count <= 1) {
         readHolds.remove();
         // 如果沒有獲取鎖卻釋放則會報該錯誤
         if (count <= 0)
           throw unmatchedUnlockException();
       }
       --rh.count;
     }
     // 操作2：循環(huán)中利用CAS修改讀鎖狀態(tài)
     for (;;) {
       int c = getState();
       int nextc = c - SHARED_UNIT;
       if (compareAndSetState(c, nextc))
         // Releasing the read lock has no effect on readers,
         // but it may allow waiting writers to proceed if
         // both read and write locks are now free.
         return nextc == 0;
     }
   }

寫鎖的獲取

清單8：寫鎖的獲取入口

// WriteLock
 public void lock() {
    sync.acquire(1);
  }
// AQS
 public final void acquire(int arg) {
    // 嘗試獲取，獲取失敗后入隊，入隊失敗則interrupt當(dāng)前線程
    if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
      selfInterrupt();
  }

寫鎖的獲取也主要是tryAcquire(arg)方法，這里也拆解步驟：

操作1：如果讀鎖數(shù)量不為0或者寫鎖數(shù)量不為0，并且不是重入操作，則獲取失敗。
操作2：如果當(dāng)前鎖的數(shù)量為0，也就是不存在操作1的情況，那么該線程是有資格獲取到寫鎖，因此修改狀態(tài)，設(shè)置獨占線程為當(dāng)前線程

清單9：寫鎖的獲取

protected final boolean tryAcquire(int acquires) {
  Thread current = Thread.currentThread();
  int c = getState();
  int w = exclusiveCount(c);
  // 操作1：c != 0，說明存在讀鎖或者寫鎖
  if (c != 0) {
    // (Note: if c != 0 and w == 0 then shared count != 0) 
    // 寫鎖為0，讀鎖不為0 或者獲取寫鎖的線程并不是當(dāng)前線程，直接失敗
    if (w == 0 || current != getExclusiveOwnerThread())
      return false;
    if (w + exclusiveCount(acquires) > MAX_COUNT)
      throw new Error("Maximum lock count exceeded");
    // Reentrant acquire
    // 執(zhí)行到這里說明是寫鎖線程的重入操作，直接修改狀態(tài)，也不需要CAS因為沒有競爭
    setState(c + acquires);
    return true;
  }
  // 操作2：獲取寫鎖，writerShouldBlock對于非公平模式直接返回fasle，對于公平模式則線程需要排隊，因此需要阻塞。
  if (writerShouldBlock() ||
    !compareAndSetState(c, c + acquires))
    return false;
  setExclusiveOwnerThread(current);
  return true;
}

寫鎖的釋放

清單10：寫鎖的釋放入口

// WriteLock
public void unlock() {
    sync.release(1);
  }
// AQS
public final boolean release(int arg) {
  // 釋放鎖成功后喚醒隊列中第一個線程
  if (tryRelease(arg)) {
    Node h = head;
    if (h != null && h.waitStatus != 0)
      unparkSuccessor(h);
    return true;
  }
  return false;
}

寫鎖的釋放主要是tryRelease(arg)方法，其邏輯就比較簡單了，注釋很詳細。

清單11：寫鎖的釋放

protected final boolean tryRelease(int releases) {
   // 如果當(dāng)前線程沒有獲取寫鎖卻釋放，則直接拋異常
   if (!isHeldExclusively())
     throw new IllegalMonitorStateException();
   // 狀態(tài)變更至nextc
   int nextc = getState() - releases;
   // 因為寫鎖是可以重入，所以在都釋放完畢后要把獨占標識清空
   boolean free = exclusiveCount(nextc) == 0;
   if (free)
     setExclusiveOwnerThread(null);
   // 修改狀態(tài)
   setState(nextc);
   return free;
 }

一些其他問題

鎖降級操作哪里體現(xiàn)？

鎖降級操作指的是一個線程獲取寫鎖之后再獲取讀鎖，然后讀鎖釋放掉寫鎖的過程。在tryAcquireShared(arg)獲取讀鎖的代碼中有如下代碼。

清單12：寫鎖降級策略

Thread current = Thread.currentThread();
      // 當(dāng)前狀態(tài)
      int c = getState();
      // 存在寫鎖，并且寫鎖不等于當(dāng)前線程時返回，換句話說等寫鎖為當(dāng)前線程時則可以繼續(xù)往下獲取讀鎖。
      if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1;

。。。。。讀鎖獲取。。。。。

那么鎖降級有什么用？答案是為了可見性的保證。在ReentrantReadWriteLock的javadoc中有如下代碼，其是鎖降級的一個應(yīng)用示例。

class CachedData {
 Object data;
 volatile boolean cacheValid;
 final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
 
 void processCachedData() {
  // 獲取讀鎖
  rwl.readLock().lock();
  if (!cacheValid) {
   // Must release read lock before acquiring write lock，不釋放的話下面寫鎖會獲取不成功，造成死鎖
   rwl.readLock().unlock();
   // 獲取寫鎖
   rwl.writeLock().lock();
   try {
    // Recheck state because another thread might have
    // acquired write lock and changed state before we did.
    if (!cacheValid) {
     data = ...
     cacheValid = true;
    }
    // Downgrade by acquiring read lock before releasing write lock
    // 這里再次獲取讀鎖，如果不獲取那么當(dāng)寫鎖釋放后可能其他寫線程再次獲得寫鎖，導(dǎo)致下方`use(data)`時出現(xiàn)不一致的現(xiàn)象
    // 這個操作就是降級
    rwl.readLock().lock();
   } finally {
    rwl.writeLock().unlock(); // Unlock write, still hold read
   }
  }

  try {
  // 使用完后釋放讀鎖
   use(data);
  } finally {
   rwl.readLock().unlock();
  }
 }
 }}

公平與非公平的區(qū)別

清單13：公平下的Sync

static final class FairSync extends Sync {
   private static final long serialVersionUID = -2274990926593161451L;
   final boolean writerShouldBlock() {
     return hasQueuedPredecessors(); // 隊列中是否有元素，有責(zé)當(dāng)前操作需要block
   }
   final boolean readerShouldBlock() {
     return hasQueuedPredecessors();// 隊列中是否有元素，有責(zé)當(dāng)前操作需要block
   }
 }

公平下的Sync實現(xiàn)策略是所有獲取的讀鎖或者寫鎖的線程都需要入隊排隊，按照順序依次去嘗試獲取鎖。

清單14：非公平下的Sync

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -8159625535654395037L;
    final boolean writerShouldBlock() {
      // 非公平下不考慮排隊，因此寫鎖可以競爭獲取
      return false; // writers can always barge
    }
    final boolean readerShouldBlock() {
      /* As a heuristic to avoid indefinite writer starvation,
      * block if the thread that momentarily appears to be head
      * of queue, if one exists, is a waiting writer. This is
      * only a probabilistic effect since a new reader will not
      * block if there is a waiting writer behind other enabled
      * readers that have not yet drained from the queue.
      */
      // 這里實際上是一個優(yōu)先級，如果隊列中頭部元素時寫鎖，那么讀鎖需要等待，避免寫鎖饑餓。
      return apparentlyFirstQueuedIsExclusive();
    }
  }

非公平下由于搶占式獲取鎖，寫鎖是可能產(chǎn)生饑餓，因此解決辦法就是提高寫鎖的優(yōu)先級，換句話說獲取寫鎖之前先占坑。

總結(jié)

以上所述是小編給大家介紹的Java 讀寫鎖實現(xiàn)原理淺析，希望對大家有所幫助，如果大家有任何疑問請給我留言，小編會及時回復(fù)大家的。在此也非常感謝大家對腳本之家網(wǎng)站的支持！

您可能感興趣的文章: