AQS源碼三視-JUC系列

AQS源碼三視-JUC系列

前兩篇文章介紹了AQS的核心同步機制，使用CHL同步隊列實現執行緒等待和喚醒，一個int值記錄資源量。為上層各式各樣的同步器實現畫好了模版，像已經介紹到的ReentrantLock，Semaphroe，CountDownLatch都是在模版基礎上實現的。花里胡哨，萬變不離其宗。

以下是第三部分的內容，嘗試寫完Condition部分，基本結束AQS源碼的學習，不過還是圍繞著一個隊列（條件隊列）來進行的。

Tips

在第一篇文章中介紹Node類的nextWaiter欄位的時候已經說明過它的一個欄位兩用，對於條件隊列只有在獨佔模式下才會有。所以關於Condition的所有實現的一個前提是獨佔模式，需要謹記在心，對於理解源碼非常重要。

隊列結構

條件隊列的數據結構：定義了firstWaiter指向頭節點，lastWaiter指向尾節點，node中nextWaiter指向後繼節點，沒有使用Node結構中的pred和next，Node中維護著waitStatus和thread欄位。所以條件隊列是一個頭尾節點有指向的單向鏈表，如下圖所示：

image-20220308213603570.png

和同步隊列不同的是它不需要單獨維護的head虛節點，節點的waitStatus只有兩種：CONDITION，CANCELLED。

ConditionObject內部方法

對於條件隊列的操作全部是在ConditionObject內部完成的，先詳細閱讀好這些方法

addConditionWaiter

向條件隊列插入新節點，本方法由await調用，await是是獲取鎖的情況下執行的，所以程式碼中不需要考慮並發情況，相對來說就簡單很多，沒有cas，沒有自旋，只是一個正常入隊操作。

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        // 【1】清理節點
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 【2】創建節點
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

  • 【1】插入的時候先判斷了下原尾節點狀態，如果不是CONDITION狀態，就認為節點已經是取消狀態，觸發unlinkCancelledWaiters清理節點。

  • 【2】創建的新節點初始化狀態是CONDITION，對於新插入的節點，lastWaiter一定是指向的，因為是隊列後面加入的，然後分兩種情況：

    • 1，如果隊列沒有節點了，那firstWaiter也得指向這個新插入的節點
      image-20220309232643432.png

    • 2，如果只要列表還有節點，那最後節點的next指向新節點

image-20220309233335526.png

unlinkCancelledWaiters

本方法只在兩種情況下會調用：

  • 條件等待（await）的時候發生中斷（cancellation occurred），執行緒被取消了，節點作為本執行緒的資源自然需要清除
  • 條件隊列插入新節點的時候檢查原尾節點已經CANCELLED狀態，插入的時候順便檢查了下發現原尾節點狀態不行，觸發清除操作
private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    // 遍歷到的最後面的有效節點
    Node trail = null;
    // 完整遍歷
    while (t != null) {
        Node next = t.nextWaiter;
        // 節點狀態非CONDITION
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            // 遍歷開始一直是CANCELLED狀態節點
            if (trail == null)
                firstWaiter = next;
            else // 中間段有CANCELLED狀態節點
                trail.nextWaiter = next;
            // 遍歷到隊列尾部
            if (next == null)
                lastWaiter = trail;
        }
        else
            // 記錄最後面的有效節點
            trail = t;
        t = next;
    }
}
關聯知識

這種清除CANCELLED狀態節點的操作在等待隊列上也是有的，可以回憶一下cancelAcquire方法，不過它是傳入指定節點，會從這個節點往前連續的CANCELLED狀態節點進行清除。再比如ThreadLocalMap上也有類似的操作，當發現過期slot就會觸發清理操作，清理操作也是執行到出現可用slot位置就會停止，都不會觸發完整遍歷清除的操作。

這裡就不同了，它是完整遍歷進行清除，沒有什麼特定的位置停止的邏輯。想像一下這種場景沒有觸發signal之前，所有條件隊列里的節點是不會被回收的，因為他們需要靜靜地等待被喚醒，喚醒的時候自然會檢查狀態，但是加入很長時間沒來喚醒，條件隊列里又有很多需要清理的節點就會浪費記憶體，所以好不容易觸發一次清理那麼就進行完整遍歷清理。如此不會在取消節點特別多的時候出現不停觸發清理操作。

程式碼分析

  • 遍歷清理的操作會有以下幾種場景：

    • 遍歷開始一直是CANCELLED狀態節點，此時需要注意firstWaiter的指向需要更新
    • 中間段有CANCELLED狀態節點，把清除的節點前後節點連接起來

    遍歷是從頭至尾開始的，注意trail欄位表示遍歷到的最後面的有效節點

    1，檢測當前節點是取消節點，先把這個節點nextWaiter設置成null，斷開和後繼節點的關聯，然後判斷trail是否為null，trail為null表示從頭遍歷到此時還沒有出現一個有效節點，也就是第一種情況，所以firstWaiter就指向next，表示隊列的開始就從next開始，因為前面的節點都是取消狀態了，如果trail不為null，那麼就是第二種情況，將取消節點前後節點關聯起來即可，而trail作為此時遍歷到最後一個有效節點，自然就是trail的nextWaiter指向到取消節點的後繼節點。如果後繼節點是null，表示隊列已經遍歷結束，那麼需要更新lastWaiter指向到trail。

    2，如果不是取消節點，trail更新，遍歷繼續即可

以下圖示有三個取消狀態節點的隊列清除操作的過程：

image-20220310000057475.png

Conditions 支援方法

AQS中有一個專門的Internal support methods for Conditions

transferForSignal

被喚醒時，把條件隊列的節點換到同步隊列的操作

final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    // 【1】節點狀態更新為初始化狀態
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
​
    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    //【2】入隊操作
    Node p = enq(node);
    int ws = p.waitStatus;
    //【3】
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}
  • 【1】先用cas修改節點狀態為初始狀態0，但是如果原狀態不是CONDITION那就會返回失敗，什麼場景會有waitStatus為CANCELLED的節點呢？注意這裡很關鍵：在await操作的時候執行release操作失敗，就會把新加入條件隊列的節點狀態改為CANCELLED，並沒有直接從調節隊列中清除，這種節點是無效節點，這裡遇到無效節點就會返回false，調用本方法的程式碼可以根據返回結果判斷是否繼續向前遍歷條件隊列，找到有效節點。
  • 【2】，【3】，我們知道single操作是在持有鎖的情況下進行的，操作完成後正常都會進行unLock操作，那麼也就是說single操作只需要做一件事那就是把條件隊列中可用的節點轉移到同步隊列中即可，所以當我們執行完enq方法已經完成任務。而實際程式碼做了更多：enq方法結束意味著入隊成功，方法返回的是新入隊的node的前繼節點，然後根據前繼節點的狀態判斷出是否需要進行喚醒當前節點執行緒。如果前繼節點是CANCELLED狀態或者狀態不能順利修改成SIGNAL，就會喚醒這個節點的執行緒。假如觸發了喚醒，此時的喚醒的執行緒是等待在await方法中的unpak操作上的，當喚醒的時候也應該執行那邊的程式碼，並不是像一個普通的同步隊列中節點執行緒喚醒時執行的程式碼一樣，這部分程式碼在後續分析到喚醒部分的時候描述。

isOnSyncQueue

判斷節點是否在同步隊列中，這個判斷後面方法經常需要用到，因為Condition的場景中需要兩個隊列節點的轉換操作，其中有並發的操作的情況需要考慮。

/**
 * Returns true if a node, always one that was initially placed on
 * a condition queue, is now waiting to reacquire on sync queue.
 * @param node the node
 * @return true if is reacquiring
 */
final boolean isOnSyncQueue(Node node) {
    // 【1】條件隊列判斷
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    // 【2】同步隊列判斷
    if (node.next != null) // If has successor, it must be on queue
        return true;
    /*
     * node.prev can be non-null, but not yet on queue because
     * the CAS to place it on queue can fail. So we have to
     * traverse from tail to make sure it actually made it.  It
     * will always be near the tail in calls to this method, and
     * unless the CAS failed (which is unlikely), it will be
     * there, so we hardly ever traverse much.
     */
    // 【3】遍歷
    return findNodeFromTail(node);
}
private boolean findNodeFromTail(Node node) {
  Node t = tail;
  for (;;) {
    if (t == node)
      return true;
    if (t == null)
      return false;
    t = t.prev;
  }
}
  • 【1】，在條件隊列的判斷是如果waitStatus只要還是CONDITION或者node.prev為空那麼就一定還在條件隊列，在進入同步隊列時node.prev會被先設置，所以通過node.prev判斷是不為空不能保證一定在同步隊列，但是為空就一定不在同步隊列。
  • 【2】，在同步隊列的判斷是node.next不為空就確定一定在同步隊列，我們在已經知道enq中進行節點入隊時，入隊完成的最後一步就是設置next。
  • 【3】，遍歷保底，經過前面兩個判斷的過濾，執行到這裡的情況就是node.prev != null並且node.next == null，那就是入隊入到一半的情況，所以進行一次從尾節點向前遍歷找這個節點，確保節點是否真的已經入隊到同步隊列中。我們再思考下入是尾節點cas指向還沒執行或執行失敗的情況，那麼遍歷也是找不到這個節點在同步隊列的，還不是認為沒有在同步隊列嗎？的確，所以在此划出關鍵點：看一個隊列是否進入到同步隊列就看尾節點cas指向是否成功。 另外我們可以發現所有使用到isOnSyncQueue的地方都是while循環

transferAfterCancelledWait

final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        enq(node);
        return true;
    }
    /*
     * If we lost out to a signal(), then we can't proceed
     * until it finishes its enq().  Cancelling during an
     * incomplete transfer is both rare and transient, so just
     * spin.
     */
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;
}

這個方法的邏輯：

  • 1，先判斷是否能通過CAS把狀態從CONDITION改為初始狀態，如果可以，說明這個節點還在條件隊列上，就直接執行enq，把節點遷移到同步隊列上去然後返回true。
  • 2，如果不能通過CAS把狀態從CONDITION改為初始狀態，說明節點狀態已經不是CONDITION，然後自旋判斷節點是否還沒有進入同步隊列，如果是就讓出CPU等待一下，直到節點進入到同步隊列，然後返回false

單單從這個程式碼邏輯上看是有點奇怪的，但是從調用的上層看是合理的，在await方法的分析中會涉及到。

fullyRelease

完全釋放當前state的值，也就是此時state的值是多少就調用release方法的時候傳多少。

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}
  • 這個方法也是在await系列方法里調用，在await的時候會釋放鎖，前幾篇文章已經提及過ReentrantLock的可重入特性，就是每次獲取鎖時將state累加，當await的時候一次性全部釋放才行。
  • 調用release有可能發生失敗返回false，會進入else後拋出IllegalMonitorStateException，另外release方法也可能直接拋出異常，比如ReentrantLock的實現里判斷不是持有鎖的執行緒就會拋出一樣的IllegalMonitorStateException。這都會使failed不能更新為false，所以在finally程式碼塊中會把節點的狀態改為CANCELLED。

await和signal

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    //【1】新增一個節點入隊條件隊列
    Node node = addConditionWaiter();
    // 釋放鎖，如果釋放失敗，節點狀態會變更為取消
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    //【2】
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        //【3】喚醒執行緒後
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}
private int checkInterruptWhileWaiting(Node node) {
  return Thread.interrupted() ?
    (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
  0;
}
  • 【1】首先向條件隊列加入新節點，成功後釋放鎖。
  • 【2】自選判斷節點不在同步隊列中，然後執行緒就進入等待狀態了，直到有執行緒中斷或者signal喚醒。後面喚醒後執行的程式碼就在signal方法中分析。
public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}
private void doSignal(Node first) {
  do {
    if ( (firstWaiter = first.nextWaiter) == null)
      lastWaiter = null;
    first.nextWaiter = null;
  } while (!transferForSignal(first) &&
           (first = firstWaiter) != null);
}

  • 喚醒操作的程式碼邏輯是這樣的：

    • 1，先把first.nextWaiter置為null就是把firstWaiter從條件隊列中脫離，
    • 2，在transferForSignal方法中判斷這個節點為有效節點，如果是就更新節點狀態為初始狀態，然後調用enq方法把節點放入同步隊列的尾部，然後再把前節點狀態改為SIGNAL。
    • 3，在transferForSignal方法中判斷這個節點為無效節點，就繼續從條件隊列中脫離出一個firstWaiter節點再執行transferForSignal進行判斷直到一個有效節點出現或隊列遍歷結束。

現在我們回頭去看await方法中park位置的程式碼，繼續分析喚醒後執行的程式碼邏輯：

首先我們需要清楚，一個park的執行緒喚醒起來有兩種情況：

  • 1，正常的signal操作觸發
  • 2，執行緒interrupt觸發

而我們是不清楚這個喚醒是哪個原因觸發的，舉個例子，當signal操作觸發後，執行緒也出現了interrupt，我們通過中斷標記難道就說是interrupt原因導致的unPark的嗎？肯定是不準確的。隨意就有了一下幾種情況：

  • 1，沒有中斷標記，那就不用想了，肯定是signal操作觸發

  • 2，有中斷標記，判斷這個中斷標記是signal操作前，還是signal操作後

    • 如果是signal操作前，算interrupt觸發
    • 如果是signal操作後，算signal觸發

對應著一個欄位interruptMode標記來區分這三種情況，在checkInterruptWhileWaiting方法中的返回邏輯：

  • 0

  • 有中斷標記

    • REINTERRUPT（1）
    • THROW_IE（-1）

那麼怎麼判斷中斷是signal操作前還是後呢？關鍵方法是transferAfterCancelledWait，根據這個方法的邏輯，被喚醒的執行緒的節點狀態能夠從CONDITION改成0，那意味著還沒有出發到signal，那就是interrupt觸發，就會把節點轉移到同步隊列上去，標記THROW_IE。如果狀態已經不是CONDITION，那麼自旋保證節點轉移到同步隊列成功，標記REINTERRUPT。

執行緒喚醒後退出自旋後執行以下程式碼：

// 【1】
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
    interruptMode = REINTERRUPT;
// 【2】
if (node.nextWaiter != null) // clean up if cancelled
    unlinkCancelledWaiters();
//【3】
if (interruptMode != 0)
    reportInterruptAfterWait(interruptMode);
​
private void reportInterruptAfterWait(int interruptMode)
  throws InterruptedException {
  if (interruptMode == THROW_IE)
    throw new InterruptedException();
  else if (interruptMode == REINTERRUPT)
    selfInterrupt();
}
  • 【1】，執行acquireQueued方法，會嘗試獲取一次鎖，沒獲取到就進同步隊列，執行緒再進入等待，獲取到鎖的情況返回true，表示在隊列中等待後再獲取鎖並且發現自己已經被標記中斷，並且不是THROW_IE標記，就會變更為REINTERRUPT。因為後續是REINTERRUPT會進行一次執行緒異常標記。這裡這樣操作的原因是，在進入acquireQueued方法返回true的場景是等待同步鎖的時候執行緒被標記中斷，這個時候前面執行的時候interruptMode可能是0，那麼就需要補一次執行selfInterrupt()。
  • 【2】，對條件隊列執行清理無效節點
  • 【3】，對中斷標記類型做處理，THROW_IE：拋出異常，REINTERRUPT：執行緒標記異常（selfInterrupt()）

總結

1，在Lock替換synchronized方法和語句的地方，Condition替換Object監視方法的使用。對於condition的使用，有很多場景，像LinkedBlockingQueue中就有很典型的使用，所以理解清楚這部分的實現，有助於後續閱讀其他源碼。

2，對於關鍵的兩個操作，await就是從同步隊列節點釋放，然後條件隊列節點加入；signal就是條件隊列節點移除，同步隊列節點加入。

Tags:

VirMach 便宜 VPS

QNews

QNews