多線程編程-- part5.1 互斥鎖之公平鎖-獲取鎖
基本概念
1.AQS：AbstractQueuedSynchronizer類
AQS是java中管理“鎖”的抽象類，鎖的許多公共方法都是在這個類中實現。AQS是獨占鎖(例如，ReentrantLock)和共享鎖(例如，Semaphore)的公共父類。
(01) 獨占鎖 -- 鎖在一個時間點只能被一個線程鎖占有。根據鎖的獲取機制，它又劃分為“公平鎖”和“非公平鎖”。公平鎖，是按照通過CLH等待線程按照先來先得的規則，公平的獲取鎖；而非公平鎖，則當線程要獲取鎖時，它會無視CLH等待隊列而直接獲取鎖。獨占鎖的典型實例子是ReentrantLock，此外，ReentrantReadWriteLock.WriteLock也是獨占鎖。
2.CLH隊列：Craig, Landin, and Hagersten lock queue
CLH隊列是AQS中“等待鎖”的線程隊列。在多線程中，為了保護競爭資源不被多個線程同時操作而起來錯誤，我們常常需要通過鎖來保護這些資源。在獨占鎖中，競爭資源在一個時間點只能被一個線程鎖訪問；而其它線程則需要等待。CLH就是管理這些“等待鎖”的線程的隊列。
自旋鎖：但是自旋鎖不會引起調用者睡眠，如果自旋鎖已經被別的執行單元保持，調用者就一直循環在那裏看是否該自旋鎖的保持者已經釋放了鎖
3.CAS函數：Compare And Swap
CAS函數，是比較並交換函數，它是原子操作函數；即，通過CAS操作的數據都是以原子方式進行的。例如，compareAndSetHead(), compareAndSetTail(), compareAndSetNext()等函數。它們共同的特點是，這些函數所執行的動作是以原子的方式進行的。
ReentrantLock數據結構
從圖中可以看出：
(01) ReentrantLock實現了Lock接口。
(02) ReentrantLock與sync是組合關系。ReentrantLock中，包含了Sync對象；而且，Sync是AQS的子類；更重要的是，Sync有兩個子類FairSync(公平鎖)和NonFairSync(非公平鎖)。ReentrantLock是一個獨占鎖，至於它到底是公平鎖還是非公平鎖，就取決於sync對象是"FairSync的實例"還是"NonFairSync的實例"。
獲取公平鎖
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()來自己給自己產生一個中斷。至於為什麽要自己給自己產生一個中斷，後面再介紹。
一.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源碼
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()
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()
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()在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獲取同步鎖。
<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()的作用就是當前線程自己產生一個中斷。
多線程編程-- part5.1 互斥鎖之公平鎖-獲取鎖