java 併發程式設計-AQS原始碼分析

摘要： AQS全稱是 AbstractQueuedSynchronizer （抽象佇列同步器），是通過一個先進先出的佇列（儲存等待的執行緒）來實現同步器的一個框架是一個抽象類，是java.util.concurrent包下很多多執行緒工具類的實現基礎。Lock、CountDownLatch、...

• AQS全稱是 AbstractQueuedSynchronizer （抽象佇列同步器），是通過一個先進先出的佇列（儲存等待的執行緒）來實現同步器的一個框架是一個抽象類，是java.util.concurrent包下很多多執行緒工具類的實現基礎。Lock、CountDownLatch、Semaphore等都是基於AQS實現的。所以如果想研究Lock、CountDownLatch、Semaphore等基於AQS實現的類的原始碼，明白AQS原理是很重要的一步。

AQS實現

• AQS支援兩種鎖一種是獨佔鎖（獨佔模式），一種是共享鎖（共享模式）

  • 獨佔鎖：比如像ReentrantLock就是一種獨佔鎖模式，多個執行緒去同時搶一個鎖，只有一個執行緒能搶到這個鎖，其他執行緒就只能阻塞等待鎖被釋放後重新競爭鎖。
  • 共享鎖：比如像讀寫鎖裡面的讀鎖，一個鎖可以同時被多個執行緒擁有（多個執行緒可以同時擁有讀鎖），再比如Semaphore 設定一個資源數目（可以理解為一個鎖能同時被多少個執行緒擁有）。

• ps:共享鎖跟獨佔鎖可以同時存在，比如比如讀寫鎖，讀鎖寫鎖分別對應共享鎖和獨佔鎖

• 先來介紹一下AQS的幾個主要成員變數

//AQS等待佇列的頭結點，AQS的等待佇列是基於一個雙向連結串列來實現的，這個頭結點並不包含具體的執行緒是一個空結點（注意不是null）
private transient volatile Node head;
//AQS等待佇列的尾部結點
private transient volatile Node tail;
//AQS同步器狀態，也可以說是鎖的狀態，注意volatile修飾證明這個變數狀態要對多執行緒可見
private volatile int state;
複製程式碼

• AQS的內部類Node
  • Node顧名思義就是接點的意思，前面說過AQS等待佇列是一個雙鏈表，每個執行緒進入AQS的等待佇列的時候都會被包裝成一個Node節點

  static final class Node {
  //下面兩個屬性都是說明這個結點是共享模式還是獨佔模式，具體怎麼表示下面會分析
  static final Node SHARED = new Node();
  static final Node EXCLUSIVE = null;
  
  //下面這四個屬性就是說明結點的狀態
  static final int CANCELLED =1;//由於超時或中斷，節點已被取消
  static final int SIGNAL= -1;//表示下一個節點是通過park阻塞的，需要通過unpark喚醒
  static final int CONDITION = -2;//表示執行緒在等待條件變數（先獲取鎖，加入 
  到條件等待佇列，然後釋放鎖，等待條件變數滿足條件；只有重新獲取鎖之 
  後才能返回）
  static final int PROPAGATE = -3;//表示後續結點會傳播喚醒的操作，共享模式下起作用
  
  volatile int waitStatus;
  //前驅結點（雙鏈表）
  volatile Node prev;
  //後繼結點（雙鏈表）
  volatile Node next;
  //結點所包裝的執行緒
  volatile Thread thread;
  //對於Condtion表示下一個等待條件變數的節點；其它情況下用於區分共享模式和獨佔模式
  Node nextWaiter;
  
  final boolean isShared() {
  return nextWaiter == SHARED;
  }
  
  //取得前驅結點
  final Node predecessor() throws NullPointerException {
  Node p = prev;
  if (p == null)
  //null的時候丟擲異常
  throw new NullPointerException();
  else
  return p;
  }
  
  Node() {
  }
  
  Node(Thread thread, Node mode) {
  this.nextWaiter = mode;
  this.thread = thread;
  }
  
  Node(Thread thread, int waitStatus) {
  this.waitStatus = waitStatus;
  this.thread = thread;
  }
  }
  複製程式碼

獨佔模式具體分析

acquire方法分析

• 首先是acquire方法

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

• acquire方法簡要執行說明
  • 首先是tryAcquire方法會嘗試的獲取一下鎖，成功的話就直接證明獲取鎖成功，acquire方法執行完畢（注意tryAcquire方法需要實現的子類根據自己的需要實現怎麼搶鎖，AQS不實現可以看到,只是丟擲了異常，後面我們分析一個ReentrantLock的tryAcquire給大家感受下，暫時你只需要知道這個方法只是嘗試獲取鎖）

  protected boolean tryAcquire(int arg) {
  throw new UnsupportedOperationException();
  }
  複製程式碼

  • tryAcquire方法搶鎖失敗就執行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
  • 先是addWaiter(Node.EXCLUSIVE)方法，這個方法表示將當前搶鎖執行緒包裝成結點並加入等待佇列，返回包裝後的結點
  • addWaiter方法返回的結點，作為acquireQueued方法的引數，該方法主要是等待佇列順序獲取資源
  • 注意acquireQueued返回true表示執行緒發生中斷，這時就會執行selfInterrupt方法響應中斷。
  • 由於tryAcquireAQS沒有具體實現，下面我們就接著看下addWaiter這個方法

addWaiter方法分析

private Node addWaiter(Node mode) {
//首先把當前競爭鎖的執行緒包裝成一個節點
Node node = new Node(Thread.currentThread(), mode);
//如果以前的尾結點不為null（為null表示當前結點就是等待佇列的第一個結點），
就將當前結點設定為尾結點，並通過cas操作更新tail尾新入隊的結點
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//cas更新tail失敗就以自旋的方式繼續嘗試入佇列
enq(node);
return node;
}
複製程式碼

• 繼續往下看下enq方法

private Node enq(final Node node) {
//死迴圈進行自旋操作
for (;;) {
Node t = tail;
//這裡利用了延遲載入，尾結點為空的時候生成tail結點，初始化
if (t == null) { 
//佇列為空的時候自然尾結點就等於頭結點，所以通過cas操作設定tail = head
if (compareAndSetHead(new Node()))
tail = head;
} else {
//尾結點初始化成功後就一直自旋的更新尾結點為當前結點
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
複製程式碼

• 總結一下addWaiter方法還是比較簡單的，主要是將當前競爭鎖的執行緒包裝成為Node結點然後通過先嚐試失敗在自旋的方式加入到等待佇列的尾部，同時更新尾結點tail
• addWaiter執行完畢之後就證明結點已經成功入隊，所以就要開始執行acquireQueued方法進行資源的獲取。

acquireQueued方法分析

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//獲取結點的前驅結點
final Node p = node.predecessor();
/**
如果結點的前驅結點是head表示當前結點就是等待佇列的第一個，
因為head結點並不指向一個實際的執行緒，所以這個時候就會執行下
tryAcquire函式嘗試性的獲取下鎖。因為這個時候很有可能競爭成功
**/
if (p == head && tryAcquire(arg)) {
/**
拿到鎖之後就更新頭結點為當前結點（這個結點的執行緒已經拿到鎖了，
所以更新為頭結點也不會繼續參與鎖競爭，再次提示頭結點不會參加競爭）
**/
setHead(node);
//設定以前的頭結點不指向其他結點，幫助gc
p.next = null; 
failed = false;
return interrupted;
}
/**
上面前驅不是頭結點或者獲取鎖失敗就會執行shouldParkAfterFailedAcquire
函式判斷是否應該掛起執行緒，注意只是判斷並不會執行掛起執行緒的操作，掛起執行緒的
操作由後面的parkAndCheckInterrupt函式執行
**/
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
//當出現異常的時候會執行cancelAcquire方法，來取消當前結點並從等待佇列中清除出去
if (failed)
cancelAcquire(node);
}
}
複製程式碼

• 來一起看下shouldParkAfterFailedAcquire函式

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//獲取下前驅結點的waitStatus，這個決定著是否要對後續結點進行掛起操作
int ws = pred.waitStatus;
/**
如果ws的waitStatus=-1時，證明他的後繼結點已經被park阻塞了後面到了競爭的時候會unpark喚醒後繼結
點，所以如果結點的前驅結點waitStatus是-1，shouldParkAfterFailedAcquire就會判斷需要park當前執行緒 
所以返回true。
**/
if (ws == Node.SIGNAL)
return true;
//ws>0證明前驅結點已經被取消所以需要往前找到沒有被取消的結點ws>0
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/**
前驅結點生成時，ws=0，所以如果是第一次執行shouldParkAfterFailedAcquire函式就會發現前驅結點
的ws = 0就會因為需要阻塞後面的結點設定為-1。
**/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
/**
注意把前驅結點ws設定為-1之後會雖然返回false，不掛起當前執行緒，注意只是這一次迴圈不掛起，因為
acquireQueued函式是一個死迴圈，所以到下一個迴圈如果前驅結點不是head並且tryAcquire競爭鎖失敗還 
是會執行shouldParkAfterFailedAcquire方法，這個時候前驅結點已經為-1，所以就會直接返回true
**/
return false;
}
複製程式碼

• parkAndCheckInterrupt方法主要是park阻塞執行緒並在unpark的時候返回中斷狀態

private final boolean parkAndCheckInterrupt() {
//park掛起執行緒
LockSupport.park(this);
/**
執行緒被unpark喚醒的時候會檢查終端狀態並返回，這個終端狀態會在acquireQueued方法中最後返回，
所以acquireQueued函式並不響應中斷而是返回中斷狀態由外層函式處理。
**/
return Thread.interrupted();
}
複製程式碼

• cancelAcquire函式

private void cancelAcquire(Node node) {
if (node == null)
return;
node.thread = null;
Node pred = node.prev;
//迴圈往前找到沒有被取消的結點，直到找到正常狀態的結點
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;

Node predNext = pred.next;
//因為要取消當前結點所以修改當前結點得ws為CANCELLED
node.waitStatus = Node.CANCELLED;
//如果node為尾結點就修改尾結點並將尾結點得next設為null
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
//如果不是尾結點
/**
滿足下面三個條件，將pred的next指向node的下一節點
1.pred不是head節點：如果pred為頭節點，
而node又被cancel,則node.next為等待佇列中的第一個節點，需要unpark喚醒
2.pred節點狀態為SIGNAL或能更新為SIGNAL
3.pred的thread變數不能為null
**/
int ws;
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread != null) {
Node next = node.next;
// 
if (next != null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
//如果pred為頭節點，則喚醒node的後節點,注意unparkSuccessor方法為喚醒當前結點得下一個結點
unparkSuccessor(node);
}
node.next = node; // help GC
}
}
複製程式碼

• cancleAcquire函式主要是取消當前結點，將當前結點從等待佇列中移出

• 同時遍歷前面的結點將被取消的結點從佇列中清除出去

• unparkSuccessor 方法

private void unparkSuccessor(Node node) {

int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);

Node s = node.next;
//如果下一個結點為null或者ws為取消狀態就未開始遍歷找到正常狀態的結點
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
//通過LockSupport.unpark()方法喚醒阻塞的執行緒，注意被阻塞的執行緒從哪開始繼續執行
LockSupport.unpark(s.thread);
}
複製程式碼

• 簡單回顧下整個過程
  • 首先會通過addWaiter方法來將第一次競爭失敗的執行緒包裝成Node結點自旋的方式加入到等待佇列
  • 加入到等待佇列之後就會該結點就會執行acquireQueued方法開始同等待佇列的其他結點一起獲取鎖
  • 先是判斷該結點是不是第一個實際的等待的結點（不是head結點，head結點是空節點），如果是就用tryAcquire方法嘗試獲取鎖，成功就更新head結點。
  • 如果上面的操作失敗，就會判斷該執行緒是否需要被掛起（當前驅結點的ws為signal就會被掛起），然後就掛起該執行緒，當被喚醒之後就繼續重複上面的步驟獲取鎖
  • 當獲取到鎖以後就會有一個釋放鎖，釋放鎖的方法主要是release方法

release方法

public final boolean release(int arg) {
//首先是執行tryRelease()方法，主要如何去釋放獲取到的鎖，這個方法需要子類自己去實現
if (tryRelease(arg)) {
//釋放成功以後如果發現等待佇列還有在等待獲取鎖的Node就用unparkSuccessor喚醒頭結點的下一個結點
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
//tryRelease方法呼叫失敗會返回false
return false;
}
複製程式碼

• release方法邏輯還是比較簡單的，釋放掉獲取的鎖之後喚醒等待佇列的後續Node。

• 到這裡我們獨佔模式下的AQS獲取鎖的過程就分析完了。基本上了解了獨佔模式得鎖競爭過程，在看共享模式下的鎖競爭過程就比較簡單了，看的時候注意對比著看。

共享模式具體分析

• 這裡我建議大家在看下文章前面我講的共享模式與獨佔模式得區別

首先是acquireShared方法（對應獨佔模式得acquire方法）

//注意方法會忽略中斷，沒有selfInterrupt這個方法來響應中斷
 public final void acquireShared(int arg) {
/**
這個tryAcquireShared方法對應獨佔模式的tryAcquire方法，也是需要子類自己去實現的。嘗試獲取資源。負數表示失敗；0表示成功，但沒有剩餘可用資源；正數表示成功，且有剩餘資源。<0就表示獲取失敗就進doAcquireShared方法來開始進入等待佇列等待獲取資源
**/
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
複製程式碼

doAcquireShared方法

private void doAcquireShared(int arg) {
//構造一個共享模式得結點加入到等待佇列
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head) {
//如果當前結點是除了head得佇列中的第一個結點那麼就嘗試獲取資源
int r = tryAcquireShared(arg);
//tryAcquireShared返回值>=0資源獲取成功，就開始進行更新結點操作
if (r >= 0) {
//這裡注意下獨佔模式呼叫的是setHead方法，但是共享模式呼叫的是setHeadAndPropagate方法
setHeadAndPropagate(node, r);
p.next = null; // help GC
//這裡注意下，這裡響應中斷，跟獨佔模式不同
if (interrupted)
selfInterrupt();
failed = false;
return;
}
} 
//獲取資源失敗就開始繼續判斷是否需要掛起執行緒，與獨佔模式得相同
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
//出現異常就取消結點
if (failed)
cancelAcquire(node);
}
}
複製程式碼

• 可以看到共享模式下的鎖競爭同非共享模式下的步驟大體上相同
• tryAcquireShared不同的是他會返回三種狀態，負數表示失敗；0表示成功，但沒有剩餘可用資源；正數表示成功，且有剩餘資源。因為前面說過共享模式下資源是有很多的（或者說是有多個鎖），允許最多由對應數量的執行緒持有相應的資源，一個執行緒持有一個資源量就-1直到0。
• 還有就是共享模式下的setHeadAndPropagate方法，下面一起來看下

setHeadAndPropagate方法

private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; 
//首先更新head結點
setHead(node);
/** 
注意propagate表示的上次執行tryAcquireShared方法後的返回值。>0表示還有剩餘資源，既然有剩餘資源就繼續喚醒後面等待獲取資源的並且是共享模式得Node
或者h == null
或者當前獲取到資源的得結點<0，signal需要喚醒後續結點
**/
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
//喚醒後續共享模式得Node結點
doReleaseShared();
}
}
複製程式碼

• setHeadAndPropagate方法主要乾了兩件事

  • 1. 更新頭結點
  • 2.檢查是否需要喚醒後續結點，滿足上面說的三個條件

• 看下doReleaseShared方法

private void doReleaseShared() {
for (;;) {
Node h = head;
//如果頭結點不為空，並且不是tail，佇列還有結點
if (h != null && h != tail) {
int ws = h.waitStatus;
//如果head結點ws為signal就更新為0並喚醒後續結點
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
if (h == head)
break;
}
}
複製程式碼

• 可以看到doReleaseShared方法主要就是喚醒阻塞佇列中等待的結點
• 這裡引入另外一篇博文的分析（地址：ofollow,noindex">www.jianshu.com/p/c244abd58…

  • 從上面的分析可以知道，獨佔模式和共享模式的最大區別在於獨佔模式只允許一個執行緒持有資源，而共享模式下，當呼叫doAcquireShared時，會看後續的節點是否是共享模式，如果是，會通過unpark喚醒後續節點； 從前面的分析可以知道，被喚醒的節點是被堵塞在doAcquireShared的parkAndCheckInterrupt方法，因此喚醒之後，會再次呼叫setHeadAndPropagate,從而將等待共享鎖的執行緒都喚醒，也就是說會將喚醒傳播下去；

  • 加入同步佇列並阻塞的節點，它的前驅節點只會是SIGNAL，表示前驅節點釋放鎖時，後繼節點會被喚醒。shouldParkAfterFailedAcquire()方法保證了這點，如果前驅節點不是SIGNAL,它會把它修改成SIGNAL。 造成前驅節點是PROPAGATE的情況是前驅節點獲得鎖時，會喚醒一次後繼節點，但這時候後繼節點還沒有加入到同步佇列，所以暫時把節點狀態設定為PROPAGATE,當後繼節點加入同步佇列後，會把PROPAGATE設定為SIGNAL,這樣前驅節點釋放鎖時會再次doReleaseShared，這時候它的狀態已經是SIGNAL了，就可以喚醒後續節點了。（補充下，想一下如果不考慮，沒有後繼結點的時候直接講ws置為signal，那麼每次doReleaseShared執行的之後就直接unparkSuccessor喚醒後繼結點那麼就沒意義，因為沒有後繼結點。所以在沒有後繼節點的時候ws = 0，那麼就先ws置為PROPAGATE，反正後繼結點加入的時候shouldParkAfterFailedAcquire會將前面的結點的ws置為signal）

  • 舉例說明:例如讀寫鎖，寫讀操作和寫寫操作互斥，讀讀之間不互斥；當呼叫acquireShared獲取讀鎖時，會檢查後續節點是否是獲取讀鎖，如果是，則同樣釋放；

總結

• 看完整篇博文之後各位應該就對aqs實現有一定了解了，第一次看的時候獲取會很懵逼，但其實多看幾遍慢慢就懂了，看的時候注意執行緒在什麼時候阻塞，以及被喚醒後從哪裡開始執行
• 第二個就是注意共享模式和獨佔模式的區別。另外建議各位看下基於aqs框架實現的鎖的tryAcquire方法以及tryRelease方法，比如reentrantLock。原始碼比較簡單就是搶到資源該怎麼處理，釋放資源該怎麼處理。