java.util.concurrent包中的工具實現(xiàn)核心都是AQS，了解ReentrantLock的實現(xiàn)原理，需要先分析AQS以及AQS與ReentrantLock的關(guān)系。

這篇文章中分析了ReentrantLock#lock與ReentrantLock#unlock的實現(xiàn)，對于Condition的實現(xiàn)分析，另外文章再講，基本上大同小異。

ReentrantLock實現(xiàn)核心CAQS(AbstractQueuedSynchronizer)

AQS，隊列同步器，在juc包中的工具類都是依賴于AQS來實現(xiàn)同步控制，看一張AQS的結(jié)構(gòu)圖。

同步控制中主要使用到的信息如上圖所示。AQS可以被當(dāng)做是一個同步監(jiān)視器的實現(xiàn)，并且具有排隊功能。當(dāng)線程嘗試獲取AQS的鎖時，如果AQS已經(jīng)被別的線程獲取鎖，那么將會新建一個Node節(jié)點，并且加入到AQS的等待隊列中，這個隊列也由AQS本身自己維護(hù)。當(dāng)鎖被釋放時，喚醒下一個節(jié)點嘗試獲取鎖。

• 變量exclusiveOwnerThread在互斥模式下，表示當(dāng)前持有鎖的線程。
• 變量tail指向等待獲取AQS的鎖的節(jié)點隊列的最后一個
• 變量head指向隊列中head節(jié)點，head節(jié)點信息為空，不表示任何正在等待的線程。
• 變量state表示AQS同步器的狀態(tài)，在不同情況下含義可能不太一樣，例如以下幾種
  • 在ReentrantLock中，表示AQS的鎖是否已經(jīng)被占用獲取，0：沒有，>=1：已被獲取,當(dāng)大于1時表示被同一線程多次重入鎖。
  • 在CountDownLatch中，表示計數(shù)還剩的次數(shù)，當(dāng)?shù)竭_(dá)0時，喚醒等待線程。
  • 在Semaphore中，表示AQS還可以被獲取鎖的次數(shù)，獲取一次就減1，當(dāng)?shù)竭_(dá)0時，嘗試獲取的線程將會阻塞。

Node結(jié)構(gòu)

Node節(jié)點是AQS管理的等待隊列的節(jié)點元素，除了head節(jié)點之外，其他一個節(jié)點就代表一個正在等待線程的隊列。Node一般的重要參數(shù)有幾個。

• prev 前置節(jié)點
• next后置節(jié)點
• thread 代表的線程
• waitStatus節(jié)點的等待狀態(tài)
  • 1表示節(jié)點已經(jīng)取消，也就是線程可能已經(jīng)中斷，不需要再等待獲取鎖了，在后續(xù)代碼中會處理跳過waitStatus等于1的節(jié)點
  • -1表示當(dāng)前節(jié)點的后置節(jié)點代表的線程需要被掛起
  • -2表示當(dāng)前線程正在等待的是Condition鎖

ReentrantLock實現(xiàn)分析

二者關(guān)聯(lián)

ReentrantLock實現(xiàn)核心是基于AQS,看下面一張圖，分析AQS與ReentrantLock的關(guān)系。

從圖中可以看出，ReentrantLock里面有最終兩個內(nèi)部類，F(xiàn)airSync和NonfairSync通過繼承AbstractQueuedSynchronizer的功能，來實現(xiàn)兩種同步互斥方案：公平鎖和非公平鎖。

在ReentrantLock中最終lock和unlock操作，都由FairSync和NonfairSync實際完成。

NonfairSync分析

下面看一個最簡單利用ReentrantLock實現(xiàn)互斥的例子。

      ReentrantLock lock = new ReentrantLock();
		//嘗試獲取鎖
      lock.lock();
      //獲得鎖后執(zhí)行邏輯......
      //......
		//解鎖
      lock.unlock();

在ReentrantLock的默認(rèn)無參構(gòu)造方法中，創(chuàng)建的是非公平鎖

  public ReentrantLock() {
      sync = new NonfairSync();
  }

下面分析lock.lock();這句代碼是如何實現(xiàn)同步互斥的。

NonfairSync#lock

點開lock.lock();源碼，可以看到最終實際調(diào)用的是NonfairSync#lock，這是分析的入口。

NonfairSync#lock源碼如下。

final void lock() {
  if (compareAndSetState(0, 1))//【step1】
      setExclusiveOwnerThread(Thread.currentThread());//【step2】
  else
      acquire(1);//【step3】
}

【step1】上面有提到，NonfairSync繼承自AbstractQueuedSynchronizer，NonfairSync就是一個AQS，因此在步驟【1】其實就是利用CAS（一個原子性的比較并設(shè)置操作）嘗試設(shè)置AQS的state為1。如果設(shè)置成功，表示獲取鎖成功；如果設(shè)置失敗，表示state之前已經(jīng)是>=1，已經(jīng)被別的線程占用了AQS的鎖，所示無法設(shè)置state為1，稍后會把線程加入到等待隊列。

**非公平鎖與公平鎖：**對于NonfairSync非公平鎖來說，線程只要執(zhí)行l(wèi)ock請求，就會立馬嘗試獲取鎖，不會管AQS當(dāng)前管理的等待隊列中有沒有正在等待的線程，這種操作是不公平的，沒有先來后到；而稍后介紹的FairSync公平鎖，則會在lock請求進(jìn)行時，先判斷AQS管理的等待隊列中是否已經(jīng)有正在等待的線程，有的話就是不嘗試獲取鎖，直接進(jìn)入等待隊列，保證了公平性。

這一步需要熟悉的是CAS操作，分析一下compareAndSetState源碼，如下。這一步利用unsafe包的cas操作，unsafe包類似一種java留給開發(fā)者的后門，可以用來直接操作內(nèi)存數(shù)據(jù)，并且保證這個操作的原子性。在下面的代碼中，stateOffset表示state比變量的內(nèi)存偏移地址，用來尋找state變量內(nèi)存位置。整個cas操作就是找到內(nèi)存中當(dāng)前的state變量值，并且與expect期待值比較，如果跟期待值相同，那么表示這個值是可以修改的，此時就會對state值進(jìn)行更新；如果與期待值不一樣，那么將不能進(jìn)行更新操作。unsafe保證了比較與設(shè)置值的過程是原子性的，在這一步不會出現(xiàn)線程安全問題。

protected final boolean compareAndSetState(int expect, int update) {
  // See below for intrinsics setup to support this
  return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

【step2】操作是在設(shè)置state成功之后，表示當(dāng)前線程獲取AQS鎖成功，需要設(shè)置AQS中的變量exclusiveOwnerThread為當(dāng)前持有鎖的線程，做保存記錄。

【step3】當(dāng)嘗試獲取鎖失敗的時候，就需要進(jìn)行步驟3，執(zhí)行acquire,進(jìn)行再次嘗試或者線程進(jìn)入等待隊列。

AbstractQueuedSynchronizer#acquire

當(dāng)?shù)谝淮螄L試獲取鎖失敗之后，執(zhí)行【step3】acquire方法，這個方法可以講一個嘗試獲取鎖失敗的線程放入AQS管理的等待隊列進(jìn)行等待，并且將線程掛起。實現(xiàn)邏輯在AbstractQueuedSynchronizer實現(xiàn)，NonfairSync通過繼承AbstractQueuedSynchronizer獲得等待隊列管理的功能。

下面看源碼。

public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
      selfInterrupt();
}

NonfairSync#tryAcquireC鎖重入實現(xiàn)

首先，執(zhí)行tryAcquire再次嘗試一次獲取lock，tryAcquire是由子類實現(xiàn)，也就是這里調(diào)用NonfairSync#tryAcquire方法，跟蹤調(diào)用，最終執(zhí)行代碼如下。

final boolean nonfairTryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
      //如果此時state已經(jīng)變?yōu)?，再次嘗試一次獲取鎖
      if (compareAndSetState(0, acquires)) {
          setExclusiveOwnerThread(current);
          return true;
      }
  }
  else if (current == getExclusiveOwnerThread()) {
      //否則判斷當(dāng)前持有AQS的鎖的線程是不是當(dāng)前請求獲取鎖的線程，是的話就進(jìn)行鎖重入。
      int nextc = c + acquires;
      if (nextc < 0) // overflow
          throw new Error("Maximum lock count exceeded");
      setState(nextc);
      return true;
  }
  return false;
}

對于NonfairSync#tryAcquire的實現(xiàn)，首先是重新嘗試一次獲取鎖，如果還是獲取不到，就嘗試判斷當(dāng)前的是不是屬于重入鎖的情形。

對于重入鎖的情形，就需要對state進(jìn)行累加1，意思就是重入一次就對state加1。這樣子，在解鎖的時候，每次unlock就對state減一，等到state的值為0的時候，才能喚醒下一個等待線程。

如果獲取成功，返回true；否則返回false，繼續(xù)執(zhí)行下一步acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，添加一個Node節(jié)點進(jìn)入AQS管理的等待隊列。

AbstractQueuedSynchronizer#addWaiterC添加Node到等待隊列

這個方法屬于隊列管理，也是由AbstractQueuedSynchronizer默認(rèn)實現(xiàn)，NonfairSync繼承獲得。

查看addWaiter源碼實現(xiàn)。

private Node addWaiter(Node mode) {
  //新建一個Node節(jié)點，mode傳入表示當(dāng)前線程之間競爭是互斥模式
  Node node = new Node(Thread.currentThread(), mode);
  // Try the fast path of enq; backup to full enq on failure
  //嘗試添加當(dāng)前新建Node到鏈表隊列末尾
  Node pred = tail;
  if (pred != null) {
      node.prev = pred;
      //利用cas設(shè)置尾指針指向的節(jié)點為當(dāng)前線新建節(jié)點
      if (compareAndSetTail(pred, node)) {
          pred.next = node;
          return node;
      }
  }
  //當(dāng)前是空的沒有任何正在等待的線程N(yùn)ode的時候，執(zhí)行enq(node)，初始化等待隊列
  enq(node);
  return node;
}

private Node enq(final Node node) {
  for (;;) {
      Node t = tail;
      if (t == null) { // Must initialize
          //新建一個空的頭節(jié)點
          if (compareAndSetHead(new Node()))
              tail = head;
      } else {
          //跟之前一樣，利用cas這只當(dāng)前新建節(jié)點為尾節(jié)點
          node.prev = t;
          if (compareAndSetTail(t, node)) {
              t.next = node;
              return t;
          }
      }
  }
}

以上操作，完成將一個節(jié)點加入隊列操作。加入完成之后，返回這個新加入的節(jié)點Node給acquireQueued方法繼續(xù)執(zhí)行。

acquireQueued(addWaiter(Node.EXCLUSIVE), arg))C鎖競爭優(yōu)化

上面既然都完成了等待節(jié)點如隊列的操作，為什么不直接掛起線程進(jìn)入等待呢？因此這里的設(shè)計者做了一個優(yōu)化操作。acquireQueued方法其實就是為了減少線程掛起、喚醒次數(shù)而作的優(yōu)化操作。

下面看看acquireQueued的代碼實現(xiàn)。

final boolean acquireQueued(final Node node, int arg) {
  boolean failed = true;
  try {
      boolean interrupted = false;
      for (;;) {
          //獲取當(dāng)前競爭鎖的線程N(yùn)ode的前置節(jié)點
          final Node p = node.predecessor();
          //【step1】
          if (p == head && tryAcquire(arg)) {
              setHead(node);
              p.next = null; // help GC
              failed = false;
              return interrupted;
          }
          //【step2】
          if (shouldParkAfterFailedAcquire(p, node) &&
              parkAndCheckInterrupt())
              interrupted = true;
      }
  } finally {
      if (failed)
          cancelAcquire(node);
  }
}

【step1】假如前置節(jié)點是head，那么表示當(dāng)前線程是等待隊列中最大優(yōu)先級的等待線程，可以繼續(xù)最后的嘗試獲取鎖，因為很有可能會獲取到鎖，從而避免線程掛起、喚醒，耗費(fèi)資源，這里也算是最終一次嘗試獲取。

【step2】shouldParkAfterFailedAcquire(p, node)是檢查當(dāng)前是否有必要掛起，前面我們說過，只有當(dāng)前置節(jié)點的waitStatus是-1的時候才會掛起當(dāng)前節(jié)點代表的線程。當(dāng)shouldParkAfterFailedAcquire(p, node)返回true的時候，就可以執(zhí)行parkAndCheckInterrupt()來掛起線程。

shouldParkAfterFailedAcquire(p, node)源碼如下。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  int ws = pred.waitStatus;
  if (ws == Node.SIGNAL)
      //前置節(jié)點是-1，返回true表示線程可掛起
      return true;
  if (ws > 0) {
      //前置節(jié)點大于0表示前置節(jié)點已經(jīng)取消，那么進(jìn)行跳過前置節(jié)點的操作，做鏈表的基本刪除節(jié)點操作
      do {
          node.prev = pred = pred.prev;
      } while (pred.waitStatus > 0);
      pred.next = node;
  } else {
      //如果前置節(jié)點還是0,表示前置節(jié)點Node的waitStatus是初始值，需要設(shè)置為-1，然后外層循環(huán)重新執(zhí)行shouldParkAfterFailedAcquire方法，即可掛起當(dāng)前線程。
      compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  }
  return false;
}

private final boolean parkAndCheckInterrupt() {
  //阻塞掛起線程，等待喚醒
  LockSupport.park(this);
  //喚醒后，重置中斷標(biāo)記，線程中斷標(biāo)記位為不中斷
  return Thread.interrupted();
}

FairSync分析

之前提到

**非公平鎖與公平鎖：**對于NonfairSync非公平鎖來說，線程只要執(zhí)行l(wèi)ock請求，就會立馬嘗試獲取鎖，不會管AQS當(dāng)前管理的等待隊列中有沒有正在等待的線程，這種操作是不公平的，沒有先來后到；而稍后介紹的FairSync公平鎖，則會在lock請求進(jìn)行時，先判斷AQS管理的等待隊列中是否已經(jīng)有正在等待的線程，有的話就是不嘗試獲取鎖，直接進(jìn)入等待隊列，保證了公平性。

所以兩者的實現(xiàn)區(qū)別在于第一次嘗試lock的動作不一樣。

FairSync#lock

final void lock() {
  acquire(1);
}

public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
      selfInterrupt();
}

最終差異體現(xiàn)在FairSync#tryAcquire的實現(xiàn)（第一次嘗試獲取lock）

protected final boolean tryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
      //hasQueuedPredecessors判斷隊列是否還有別的線程在等待鎖，沒有的話就嘗試獲取lock
      //如果有別的線程在等待鎖，就不會嘗試獲取lock；下面如果也不是重入的情況的話就直接進(jìn)入等待隊列
      if (!hasQueuedPredecessors() &&
          compareAndSetState(0, acquires)) {
          setExclusiveOwnerThread(current);
          return true;
      }
  }
  else if (current == getExclusiveOwnerThread()) {
      int nextc = c + acquires;
      if (nextc < 0)
          throw new Error("Maximum lock count exceeded");
      setState(nextc);
      return true;
  }
  return false;
}

AbstractQueuedSynchronizer#release --AQS解鎖操作

AQS中定義了解鎖操作的模板方法，tryRelease(arg)是不同AQS子類實現(xiàn),對state的多樣化操作。例如ReentrantLock中的tryRelease(arg)操作比較明顯的就是對state減一。

tryRelease(arg)返回結(jié)果表示本次操作后是否需要喚醒下一個等待節(jié)點。對于ReentrantLock就是state減一之后是否變?yōu)?了。如果需要喚醒下一個節(jié)點的線程，那么判斷一下Head有沒有下一個要喚醒的節(jié)點線程，有的話就進(jìn)行喚醒操作unparkSuccessor(h);

public final boolean release(int arg) {
  if (tryRelease(arg)) {
      Node h = head;
      if (h != null && h.waitStatus != 0)
          unparkSuccessor(h);
      return true;
  }
  return false;
}

ReentrantLock.Sync#tryReleaseC解鎖實現(xiàn)

看一下ReentrantLock.Sync的tryRelease實現(xiàn).是如何為state減一 的。。

protected final boolean tryRelease(int releases) {
  //獲取state減掉realease，對于ReentrantLock就是默認(rèn)減一
  int c = getState() - releases;
  if (Thread.currentThread() != getExclusiveOwnerThread())
      throw new IllegalMonitorStateException();
  boolean free = false;
  if (c == 0) {
      //如果減到0了，那么久釋放鎖
      free = true;
      //設(shè)置持有線程為null
      setExclusiveOwnerThread(null);
  }
  //設(shè)置state為新的
  setState(c);
  return free;
}

至于這里設(shè)置state為啥不同cas操作，因為

  if (Thread.currentThread() != getExclusiveOwnerThread())
      throw new IllegalMonitorStateException();

所以永遠(yuǎn)只有持有鎖的線程會做解鎖減一的操作，state設(shè)置是線程安全的。

注意一下

其實這里還沒分析Condition的實現(xiàn)原理，篇幅太長，下次另開文章分析。以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持服務(wù)器之家。

原文鏈接：https://blog.csdn.net/qq_20597727/article/details/86263237