Java concurrency之公平鎖(一)_動力節點Java學院整理

基本概念

本章，我們會講解“線程獲取公平鎖”的原理；在講解之前，需要了解幾個基本概念。后面的內容，都是基于這些概念的；這些概念可能比較枯燥，但從這些概念中，能窺見“java鎖”的一些架構，這對我們了解鎖是有幫助的。
1. AQS -- 指AbstractQueuedSynchronizer類。
    AQS是java中管理“鎖”的抽象類，鎖的許多公共方法都是在這個類中實現。AQS是獨占鎖(例如，ReentrantLock)和共享鎖(例如，Semaphore)的公共父類。

2. AQS鎖的類別 -- 分為“獨占鎖”和“共享鎖”兩種。
    (01) 獨占鎖 -- 鎖在一個時間點只能被一個線程鎖占有。根據鎖的獲取機制，它又劃分為“公平鎖”和“非公平鎖”。公平鎖，是按照通過CLH等待線程按照先來先得的規則，公平的獲取鎖；而非公平鎖，則當線程要獲取鎖時，它會無視CLH等待隊列而直接獲取鎖。獨占鎖的典型實例子是ReentrantLock，此外，ReentrantReadWriteLock.WriteLock也是獨占鎖。
    (02) 共享鎖 -- 能被多個線程同時擁有，能被共享的鎖。JUC包中的ReentrantReadWriteLock.ReadLock，CyclicBarrier， CountDownLatch和Semaphore都是共享鎖。這些鎖的用途和原理，在以后的章節再詳細介紹。

3. CLH隊列 -- Craig, Landin, and Hagersten lock queue
    CLH隊列是AQS中“等待鎖”的線程隊列。在多線程中，為了保護競爭資源不被多個線程同時操作而起來錯誤，我們常常需要通過鎖來保護這些資源。在獨占鎖中，競爭資源在一個時間點只能被一個線程鎖訪問；而其它線程則需要等待。CLH就是管理這些“等待鎖”的線程的隊列。
    CLH是一個非阻塞的 FIFO 隊列。也就是說往里面插入或移除一個節點的時候，在并發條件下不會阻塞，而是通過自旋鎖和 CAS 保證節點插入和移除的原子性。

4. CAS函數 -- Compare And Swap
    CAS函數，是比較并交換函數，它是原子操作函數；即，通過CAS操作的數據都是以原子方式進行的。例如，compareAndSetHead(), compareAndSetTail(), compareAndSetNext()等函數。它們共同的特點是，這些函數所執行的動作是以原子的方式進行的。

本章是圍繞“公平鎖”如何獲取鎖而層次展開。“公平鎖”涉及到的知識點比較多，但總的來說，不是特別難；如果讀者能讀懂AQS和ReentrantLock.java這兩個類的大致意思，理解鎖的原理和機制也就不成問題了。本章只是作者本人對鎖的一點點理解，希望這部分知識能幫助您了解“公平鎖”的獲取過程，認識“鎖”的框架。

ReentrantLock數據結構

ReentrantLock的UML類圖

從圖中可以看出：

(01) ReentrantLock實現了Lock接口。
(02) ReentrantLock與sync是組合關系。ReentrantLock中，包含了Sync對象；而且，Sync是AQS的子類；更重要的是，Sync有兩個子類FairSync(公平鎖)和NonFairSync(非公平鎖)。ReentrantLock是一個獨占鎖，至于它到底是公平鎖還是非公平鎖，就取決于sync對象是"FairSync的實例"還是"NonFairSync的實例"。

獲取公平鎖(基于JDK1.7.0_40)

通過前面“Java多線程系列--“JUC鎖”02之 互斥鎖ReentrantLock”的“示例1”，我們知道，獲取鎖是通過lock()函數。下面，我們以lock()對獲取公平鎖的過程進行展開。

1. lock()

lock()在ReentrantLock.java的FairSync類中實現，它的源碼如下：

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

說明：“當前線程”實際上是通過acquire(1)獲取鎖的。

        這里說明一下“1”的含義，它是設置“鎖的狀態”的參數。對于“獨占鎖”而言，鎖處于可獲取狀態時，它的狀態值是0；鎖被線程初次獲取到了，它的狀態值就變成了1。
        由于ReentrantLock(公平鎖/非公平鎖)是可重入鎖，所以“獨占鎖”可以被單個線程多此獲取，每獲取1次就將鎖的狀態+1。也就是說，初次獲取鎖時，通過acquire(1)將鎖的狀態值設為1；再次獲取鎖時，將鎖的狀態值設為2；依次類推...這就是為什么獲取鎖時，傳入的參數是1的原因了。
        可重入就是指鎖可以被單個線程多次獲取。

2. acquire()

acquire()在AQS中實現的，它的源碼如下：

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

(01) “當前線程”首先通過tryAcquire()嘗試獲取鎖。獲取成功的話，直接返回；嘗試失敗的話，進入到等待隊列排序等待(前面還有可能有需要線程在等待該鎖)。
(02) “當前線程”嘗試失敗的情況下，先通過addWaiter(Node.EXCLUSIVE)來將“當前線程”加入到"CLH隊列(非阻塞的FIFO隊列)"末尾。CLH隊列就是線程等待隊列。
(03) 再執行完addWaiter(Node.EXCLUSIVE)之后，會調用acquireQueued()來獲取鎖。由于此時ReentrantLock是公平鎖，它會根據公平性原則來獲取鎖。
(04) “當前線程”在執行acquireQueued()時，會進入到CLH隊列中休眠等待，直到獲取鎖了才返回！如果“當前線程”在休眠等待過程中被中斷過，acquireQueued會返回true，此時"當前線程"會調用selfInterrupt()來自己給自己產生一個中斷。至于為什么要自己給自己產生一個中斷，后面再介紹。

上面是對acquire()的概括性說明。下面，我們將該函數分為4部分來逐步解析。

 一. tryAcquire()

1. tryAcquire()

公平鎖的tryAcquire()在ReentrantLock.java的FairSync類中實現，源碼如下：

protected final boolean tryAcquire(int acquires) {
 // 獲取“當前線程”
 final Thread current = Thread.currentThread();
 // 獲取“獨占鎖”的狀態
 int c = getState();
 // c=0意味著“鎖沒有被任何線程鎖擁有”，
 if (c == 0) {
  // 若“鎖沒有被任何線程鎖擁有”，
  // 則判斷“當前線程”是不是CLH隊列中的第一個線程線程，
  // 若是的話，則獲取該鎖，設置鎖的狀態，并切設置鎖的擁有者為“當前線程”。
  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;
}

說明：根據代碼，我們可以分析出，tryAcquire()的作用就是嘗試去獲取鎖。注意，這里只是嘗試！
         嘗試成功的話，返回true；嘗試失敗的話，返回false，后續再通過其它辦法來獲取該鎖。后面我們會說明，在嘗試失敗的情況下，是如何一步步獲取鎖的。

2. hasQueuedPredecessors()

hasQueuedPredecessors()在AQS中實現，源碼如下：

public final boolean hasQueuedPredecessors() {
 Node t = tail; 
 Node h = head;
 Node s;
 return h != t &&
  ((s = h.next) == null || s.thread != Thread.currentThread());
}

說明： 通過代碼，能分析出，hasQueuedPredecessors() 是通過判斷"當前線程"是不是在CLH隊列的隊首，來返回AQS中是不是有比“當前線程”等待更久的線程。下面對head、tail和Node進行說明。

 3. Node的源碼

Node就是CLH隊列的節點。Node在AQS中實現，它的數據結構如下：

private transient volatile Node head; // CLH隊列的隊首
private transient volatile Node tail; // CLH隊列的隊尾

// CLH隊列的節點
static final class Node {
 static final Node SHARED = new Node();
 static final Node EXCLUSIVE = null;

 // 線程已被取消，對應的waitStatus的值
 static final int CANCELLED = 1;
 // “當前線程的后繼線程需要被unpark(喚醒)”，對應的waitStatus的值。
 // 一般發生情況是：當前線程的后繼線程處于阻塞狀態，而當前線程被release或cancel掉，因此需要喚醒當前線程的后繼線程。
 static final int SIGNAL = -1;
 // 線程(處在Condition休眠狀態)在等待Condition喚醒，對應的waitStatus的值
 static final int CONDITION = -2;
 // (共享鎖)其它線程獲取到“共享鎖”，對應的waitStatus的值
 static final int PROPAGATE = -3;

 // waitStatus為“CANCELLED, SIGNAL, CONDITION, PROPAGATE”時分別表示不同狀態，
 // 若waitStatus=0，則意味著當前線程不屬于上面的任何一種狀態。
 volatile int waitStatus;

 // 前一節點
 volatile Node prev;

 // 后一節點
 volatile Node next;

 // 節點所對應的線程
 volatile Thread thread;

 // nextWaiter是“區別當前CLH隊列是 ‘獨占鎖'隊列 還是 ‘共享鎖'隊列 的標記”
 // 若nextWaiter=SHARED，則CLH隊列是“獨占鎖”隊列；
 // 若nextWaiter=EXCLUSIVE，(即nextWaiter=null)，則CLH隊列是“共享鎖”隊列。
 Node nextWaiter;

 // “共享鎖”則返回true，“獨占鎖”則返回false。
 final boolean isShared() {
  return nextWaiter == SHARED;
 }

 // 返回前一節點
 final Node predecessor() throws NullPointerException {
  Node p = prev;
  if (p == null)
   throw new NullPointerException();
  else
   return p;
 }

 Node() { // Used to establish initial head or SHARED marker
 }

 // 構造函數。thread是節點所對應的線程，mode是用來表示thread的鎖是“獨占鎖”還是“共享鎖”。
 Node(Thread thread, Node mode) {  // Used by addWaiter
  this.nextWaiter = mode;
  this.thread = thread;
 }

 // 構造函數。thread是節點所對應的線程，waitStatus是線程的等待狀態。
 Node(Thread thread, int waitStatus) { // Used by Condition
  this.waitStatus = waitStatus;
  this.thread = thread;
 }
}

說明：
Node是CLH隊列的節點，代表“等待鎖的線程隊列”。
(01) 每個Node都會一個線程對應。
(02) 每個Node會通過prev和next分別指向上一個節點和下一個節點，這分別代表上一個等待線程和下一個等待線程。
(03) Node通過waitStatus保存線程的等待狀態。
(04) Node通過nextWaiter來區分線程是“獨占鎖”線程還是“共享鎖”線程。如果是“獨占鎖”線程，則nextWaiter的值為EXCLUSIVE；如果是“共享鎖”線程，則nextWaiter的值是SHARED。

 4. compareAndSetState()

compareAndSetState()在AQS中實現。它的源碼如下：

protected final boolean compareAndSetState(int expect, int update) {
 return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

說明： compareAndSwapInt() 是sun.misc.Unsafe類中的一個本地方法。對此，我們需要了解的是 compareAndSetState(expect, update) 是以原子的方式操作當前線程；若當前線程的狀態為expect，則設置它的狀態為update。 

5. setExclusiveOwnerThread()

setExclusiveOwnerThread()在AbstractOwnableSynchronizer.java中實現，它的源碼如下：

// exclusiveOwnerThread是當前擁有“獨占鎖”的線程
private transient Thread exclusiveOwnerThread;
protected final void setExclusiveOwnerThread(Thread t) {
 exclusiveOwnerThread = t;
}

說明：setExclusiveOwnerThread()的作用就是，設置線程t為當前擁有“獨占鎖”的線程。 

6. getState(), setState()

getState()和setState()都在AQS中實現，源碼如下：

// 鎖的狀態
private volatile int state;
// 設置鎖的狀態
protected final void setState(int newState) {
 state = newState;
}
// 獲取鎖的狀態
protected final int getState() {
 return state;
}

說明：state表示鎖的狀態，對于“獨占鎖”而已，state=0表示鎖是可獲取狀態(即，鎖沒有被任何線程鎖持有)。由于java中的獨占鎖是可重入的，state的值可以>1。

小結：tryAcquire()的作用就是讓“當前線程”嘗試獲取鎖。獲取成功返回true，失敗則返回false。 

二. addWaiter(Node.EXCLUSIVE)

addWaiter(Node.EXCLUSIVE)的作用是，創建“當前線程”的Node節點，且Node中記錄“當前線程”對應的鎖是“獨占鎖”類型，并且將該節點添加到CLH隊列的末尾。

1.addWaiter()

addWaiter()在AQS中實現，源碼如下：

private Node addWaiter(Node mode) {
 // 新建一個Node節點，節點對應的線程是“當前線程”，“當前線程”的鎖的模型是mode。
 Node node = new Node(Thread.currentThread(), mode);
 Node pred = tail;
 // 若CLH隊列不為空，則將“當前線程”添加到CLH隊列末尾
 if (pred != null) {
  node.prev = pred;
  if (compareAndSetTail(pred, node)) {
   pred.next = node;
   return node;
  }
 }
 // 若CLH隊列為空，則調用enq()新建CLH隊列，然后再將“當前線程”添加到CLH隊列中。
 enq(node);
 return node;
}

說明：對于“公平鎖”而言，addWaiter(Node.EXCLUSIVE)會首先創建一個Node節點，節點的類型是“獨占鎖”(Node.EXCLUSIVE)類型。然后，再將該節點添加到CLH隊列的末尾。 

2. compareAndSetTail()

compareAndSetTail()在AQS中實現，源碼如下：

private final boolean compareAndSetTail(Node expect, Node update) {
 return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

說明：compareAndSetTail也屬于CAS函數，也是通過“本地方法”實現的。compareAndSetTail(expect, update)會以原子的方式進行操作，它的作用是判斷CLH隊列的隊尾是不是為expect，是的話，就將隊尾設為update。 

3. enq()

enq()在AQS中實現，源碼如下：

private Node enq(final Node node) {
 for (;;) {
  Node t = tail;
  if (t == null) { // Must initialize
   if (compareAndSetHead(new Node()))
    tail = head;
  } else {
   node.prev = t;
   if (compareAndSetTail(t, node)) {
    t.next = node;
    return t;
   }
  }
 }
}

說明： enq()的作用很簡單。如果CLH隊列為空，則新建一個CLH表頭；然后將node添加到CLH末尾。否則，直接將node添加到CLH末尾。

小結：addWaiter()的作用，就是將當前線程添加到CLH隊列中。這就意味著將當前線程添加到等待獲取“鎖”的等待線程隊列中了。 

三. acquireQueued()

前面，我們已經將當前線程添加到CLH隊列中了。而acquireQueued()的作用就是逐步的去執行CLH隊列的線程，如果當前線程獲取到了鎖，則返回；否則，當前線程進行休眠，直到喚醒并重新獲取鎖了才返回。下面，我們看看acquireQueued()的具體流程。

 1. acquireQueued()

acquireQueued()在AQS中實現，源碼如下：

final boolean acquireQueued(final Node node, int arg) {
 boolean failed = true;
 try {
  // interrupted表示在CLH隊列的調度中，
  // “當前線程”在休眠時，有沒有被中斷過。
  boolean interrupted = false;
  for (;;) {
   // 獲取上一個節點。
   // node是“當前線程”對應的節點，這里就意味著“獲取上一個等待鎖的線程”。
   final Node p = node.predecessor();
   if (p == head && tryAcquire(arg)) {
    setHead(node);
    p.next = null; // help GC
    failed = false;
    return interrupted;
   }
   if (shouldParkAfterFailedAcquire(p, node) &&
    parkAndCheckInterrupt())
    interrupted = true;
  }
 } finally {
  if (failed)
   cancelAcquire(node);
 }
}

說明：acquireQueued()的目的是從隊列中獲取鎖。

2. shouldParkAfterFailedAcquire()

shouldParkAfterFailedAcquire()在AQS中實現，源碼如下：

// 返回“當前線程是否應該阻塞”
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 // 前繼節點的狀態
 int ws = pred.waitStatus;
 // 如果前繼節點是SIGNAL狀態，則意味這當前線程需要被unpark喚醒。此時，返回true。
 if (ws == Node.SIGNAL)
  return true;
 // 如果前繼節點是“取消”狀態，則設置 “當前節點”的 “當前前繼節點” 為 “‘原前繼節點'的前繼節點”。
 if (ws > 0) {
  do {
   node.prev = pred = pred.prev;
  } while (pred.waitStatus > 0);
  pred.next = node;
 } else {
  // 如果前繼節點為“0”或者“共享鎖”狀態，則設置前繼節點為SIGNAL狀態。
  compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
 }
 return false;
}

說明：

(01) 關于waitStatus請參考下表(中擴號內為waitStatus的值)，更多關于waitStatus的內容，可以參考前面的Node類的介紹。

CANCELLED[1]  -- 當前線程已被取消
SIGNAL[-1]    -- “當前線程的后繼線程需要被unpark(喚醒)”。一般發生情況是：當前線程的后繼線程處于阻塞狀態，而當前線程被release或cancel掉，因此需要喚醒當前線程的后繼線程。
CONDITION[-2] -- 當前線程(處在Condition休眠狀態)在等待Condition喚醒
PROPAGATE[-3] -- (共享鎖)其它線程獲取到“共享鎖”
[0]           -- 當前線程不屬于上面的任何一種狀態。

(02) shouldParkAfterFailedAcquire()通過以下規則，判斷“當前線程”是否需要被阻塞。

規則1：如果前繼節點狀態為SIGNAL，表明當前節點需要被unpark(喚醒)，此時則返回true。
規則2：如果前繼節點狀態為CANCELLED(ws>0)，說明前繼節點已經被取消，則通過先前回溯找到一個有效(非CANCELLED狀態)的節點，并返回false。
規則3：如果前繼節點狀態為非SIGNAL、非CANCELLED，則設置前繼的狀態為SIGNAL，并返回false。
如果“規則1”發生，即“前繼節點是SIGNAL”狀態，則意味著“當前線程”需要被阻塞。接下來會調用parkAndCheckInterrupt()阻塞當前線程，直到當前先被喚醒才從parkAndCheckInterrupt()中返回。

 3. parkAndCheckInterrupt())

parkAndCheckInterrupt()在AQS中實現，源碼如下：

private final boolean parkAndCheckInterrupt() {
 // 通過LockSupport的park()阻塞“當前線程”。
 LockSupport.park(this);
 // 返回線程的中斷狀態。
 return Thread.interrupted();
}

說明：parkAndCheckInterrupt()的作用是阻塞當前線程，并且返回“線程被喚醒之后”的中斷狀態。
它會先通過LockSupport.park()阻塞“當前線程”，然后通過Thread.interrupted()返回線程的中斷狀態。

這里介紹一下線程被阻塞之后如何喚醒。一般有2種情況：
第1種情況：unpark()喚醒。“前繼節點對應的線程”使用完鎖之后，通過unpark()方式喚醒當前線程。
第2種情況：中斷喚醒。其它線程通過interrupt()中斷當前線程。

補充：LockSupport()中的park(),unpark()的作用 和 Object中的wait(),notify()作用類似，是阻塞/喚醒。
它們的用法不同，park(),unpark()是輕量級的，而wait(),notify()是必須先通過Synchronized獲取同步鎖。
關于LockSupport，我們會在之后的章節再專門進行介紹！

 4. 再次tryAcquire()

了解了shouldParkAfterFailedAcquire()和parkAndCheckInterrupt()函數之后。我們接著分析acquireQueued()的for循環部分。

final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
 setHead(node);
 p.next = null; // help GC
 failed = false;
 return interrupted;
}

說明：

(01) 通過node.predecessor()獲取前繼節點。predecessor()就是返回node的前繼節點，若對此有疑惑可以查看下面關于Node類的介紹。

(02) p == head && tryAcquire(arg)
       首先，判斷“前繼節點”是不是CHL表頭。如果是的話，則通過tryAcquire()嘗試獲取鎖。
       其實，這樣做的目的是為了“讓當前線程獲取鎖”，但是為什么需要先判斷p==head呢？理解這個對理解“公平鎖”的機制很重要，因為這么做的原因就是為了保證公平性！
       (a) 前面，我們在shouldParkAfterFailedAcquire()我們判斷“當前線程”是否需要阻塞；
       (b) 接著，“當前線程”阻塞的話，會調用parkAndCheckInterrupt()來阻塞線程。當線程被解除阻塞的時候，我們會返回線程的中斷狀態。而線程被解決阻塞，可能是由于“線程被中斷”，也可能是由于“其它線程調用了該線程的unpark()函數”。
       (c) 再回到p==head這里。如果當前線程是因為其它線程調用了unpark()函數而被喚醒，那么喚醒它的線程，應該是它的前繼節點所對應的線程(關于這一點，后面在“釋放鎖”的過程中會看到)。 OK，是前繼節點調用unpark()喚醒了當前線程！
            此時，再來理解p==head就很簡單了：當前繼節點是CLH隊列的頭節點，并且它釋放鎖之后；就輪到當前節點獲取鎖了。然后，當前節點通過tryAcquire()獲取鎖；獲取成功的話，通過setHead(node)設置當前節點為頭節點，并返回。
       總之，如果“前繼節點調用unpark()喚醒了當前線程”并且“前繼節點是CLH表頭”，此時就是滿足p==head，也就是符合公平性原則的。否則，如果當前線程是因為“線程被中斷”而喚醒，那么顯然就不是公平了。這就是為什么說p==head就是保證公平性！

小結：acquireQueued()的作用就是“當前線程”會根據公平性原則進行阻塞等待，直到獲取鎖為止；并且返回當前線程在等待過程中有沒有并中斷過。

 四. selfInterrupt()

selfInterrupt()是AQS中實現，源碼如下：

private static void selfInterrupt() {
 Thread.currentThread().interrupt();
}

說明：selfInterrupt()的代碼很簡單，就是“當前線程”自己產生一個中斷。但是，為什么需要這么做呢？
這必須結合acquireQueued()進行分析。如果在acquireQueued()中，當前線程被中斷過，則執行selfInterrupt()；否則不會執行。

在acquireQueued()中，即使是線程在阻塞狀態被中斷喚醒而獲取到cpu執行權利；但是，如果該線程的前面還有其它等待鎖的線程，根據公平性原則，該線程依然無法獲取到鎖。它會再次阻塞！ 該線程再次阻塞，直到該線程被它的前面等待鎖的線程鎖喚醒；線程才會獲取鎖，然后“真正執行起來”！
也就是說，在該線程“成功獲取鎖并真正執行起來”之前，它的中斷會被忽略并且中斷標記會被清除！ 因為在parkAndCheckInterrupt()中，我們線程的中斷狀態時調用了Thread.interrupted()。該函數不同于Thread的isInterrupted()函數，isInterrupted()僅僅返回中斷狀態，而interrupted()在返回當前中斷狀態之后，還會清除中斷狀態。 正因為之前的中斷狀態被清除了，所以這里需要調用selfInterrupt()重新產生一個中斷！ 

小結：selfInterrupt()的作用就是當前線程自己產生一個中斷。

總結

再回過頭看看acquire()函數，它最終的目的是獲取鎖！

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

(01) 先是通過tryAcquire()嘗試獲取鎖。獲取成功的話，直接返回；嘗試失敗的話，再通過acquireQueued()獲取鎖。
(02) 嘗試失敗的情況下，會先通過addWaiter()來將“當前線程”加入到"CLH隊列"末尾；然后調用acquireQueued()，在CLH隊列中排序等待獲取鎖，在此過程中，線程處于休眠狀態。直到獲取鎖了才返回。 如果在休眠等待過程中被中斷過，則調用selfInterrupt()來自己產生一個中斷。

 以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持億速云。