JDK源碼分析-AbstractQueuedSynchronizer(2)

時間  2021-02-14
標籤 java node 微信 app 源碼分析 this atom spa 線程 欄目 Java 简体版
原文   原文鏈接
本文首發於微信公衆號【WriteOnRead】，歡迎關注。

1. 概述

前文「JDK源碼分析-AbstractQueuedSynchronizer(1)」初步分析了 AQS，其中提到了 Node 節點的「獨佔模式」和「共享模式」，其實 AQS 也主要是圍繞對這兩種模式的操做進行的。java

Node 節點是對線程 Thread 類的封裝，所以兩種模式能夠理解以下：node

  • 獨佔模式（exclusive）：線程對資源的訪問是排他的，即某個時間只能一個線程單獨訪問資源；
  • 共享模式（shared）：與獨佔模式不一樣，多個線程能夠同時訪問資源。

本文先分析獨佔模式下的各類操做，後面再分析共享模式。微信

2. 獨佔模式

2.1 方法概述

獨佔模式下的操做主要有如下幾個方法（可與前面分析的 Lock 接口的方法類比）：app

  1. acquire(int arg)

    以獨佔模式獲取資源，忽略中斷；能夠類比 Lock 接口的 lock 方法；源碼分析

  2. acquireInterruptibly(int arg)

    以獨佔模式獲取資源，響應中斷；能夠類比 Lock 接口的 lockInterruptibly 方法；ui

  3. tryAcquireNanos(int arg, long nanosTimeout)

    以獨佔模式獲取資源，響應中斷，且有超時等待；能夠類比 Lock 接口的 tryLock(long, TimeUnit) 方法；this

  4. release(int arg)

    釋放資源，能夠類比 Lock 接口的 unlock 方法。atom

2.2 方法分析

2.2.1 獨佔模式獲取資源（忽略中斷）

這幾種獲取資源的方法不少地方是相似的。咱們先從 acquire 方法開始分析，以下：spa

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

該方法看似很短，實際上是內部作了封裝。這幾行代碼包含了以下四個操做步驟：線程

  1. tryAcquire
  2. addWaiter(Node.EXECUSIVE)
  3. acquireQueued(final Node node, arg))
  4. selfInterrupt

上面的四個步驟不必定所有執行，下面依次進行分析。

  • step 1: tryAcquire
protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

該方法的做用是嘗試以獨佔模式獲取資源，若成功則返回 true。

能夠看到該方法是一個 protected 方法，並且 AQS 中該方法直接拋出了異常，實際上是它把實現委託給了子類。這也是 ReentrantLock、CountdownLatch 等類（嚴格來講是其內部類 Sync）的實現功能不一樣的地方，這些類正是經過對該方法的不一樣實現來制定了本身的「遊戲規則」。

若 step 1 中的 tryAcquire 方法返回 true，則表示當前線程獲取資源成功，方法直接返回，該線程接下來就能夠「隨心所欲」了；不然表示獲取失敗，接下來會依次執行 step 2 和 step 3。

  • step 2: addWaiter(Node.EXECUSIVE)
private Node addWaiter(Node mode) {
    // 將當前線程封裝爲一個 Node 節點，指定 mode
    // PS: 獨佔模式 Node.EXECUSIVE, 共享模式 Node.SHARED
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        // 經過 CAS 操做設置主隊列的尾節點
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 尾節點 tail 爲 null，表示主隊列未初始化
    enq(node);
    return node;
}

enq 方法：

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        // 尾節點爲空，代表當前隊列未初始化
        if (t == null) { // Must initialize
            // 將隊列的頭尾節點都設置爲一個新的節點
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 將 node 節點插入主隊列末尾
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

能夠看到 addWaiter(Node.EXECUSIVE) 方法的做用是：把當前線程封裝成一個獨佔模式的 Node 節點，並插入到主隊列末尾（若主隊列未初始化，則將其初始化後再插入）。

  • step 3: acquireQueued(final Node node, arg))
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        // 中斷標誌位
        boolean interrupted = false;
        for (;;) {
            // 獲取該節點的前驅節點
            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);
    }
}

若當前節點的前驅節點爲頭節點，則會再次嘗試獲取資源（tryAcuqire），若獲取成功，則將當前節點設置爲頭節點並返回；不然，若前驅節點不是頭節點，或者獲取資源失敗，則執行以下兩個方法：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 前驅節點的等待狀態
    int ws = pred.waitStatus;
    // 若前驅節點的等待狀態爲 SIGNAL，返回 true，表示當前線程能夠休眠
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    // 若前驅節點的狀態大於 0，表示前驅節點處於取消（CANCELLED）狀態
    // 則將前驅節點跳過（至關於踢出隊列）
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * waitStatus must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
         // 此時 waitStatus 只能爲 0 或 PROPAGATE 狀態，將前驅節點的等着狀態設置爲 SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

該方法的流程：

  1. 若前驅節點的等待狀態爲 SIGNAL，返回 true，表示當前線程能夠休眠（park）；
  2. 若前驅節點是取消狀態 (ws > 0)，則將其清理出隊列，以此類推；
  3. 若前驅節點爲 0 或 PROPAGATE，則將其設置爲 SIGNAL 狀態。

正如其名，該方法（shouldParkAfterFailedAcquire）的做用就是判斷當前線程在獲取資源失敗後，是否能夠休眠（park）。

parkAndCheckInterrupt:

private final boolean parkAndCheckInterrupt() {
    // 將當前線程休眠
    LockSupport.park(this);
    return Thread.interrupted();
}

該方法的做用：

  1. 使當前線程休眠（park）；
  2. 返回該線程是否被中斷（其餘線程對其發過中斷信號）。

上面就是 acquireQueued(final Node node, arg)) 方法的執行過程，爲了便於理解，可參考下面的流程圖：

image.png

若此期間被其餘線程中斷過，則此時再去執行 selfInterrupt 方法去響應中斷請求：

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

以上就是 acquire 方法執行的總體流程。

2.2.2 以獨佔模式獲取資源（響應中斷）

該操做其實與前面的過程相似，所以分析相對簡單些，代碼以下：

public final void acquireInterruptibly(int arg)
        throws InterruptedException 
    // 若線程被中斷過，則拋出異常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 嘗試獲取資源
    if (!tryAcquire(arg))
        // 嘗試獲取資源失敗
        doAcquireInterruptibly(arg);
}

tryAcquire 與前面的操做同樣，若嘗試獲取資源成功則直接返回；不然，執行 doAcquireInterruptibly:

private void doAcquireInterruptibly(int arg)
    throws InterruptedException 
    // 將當前線程封裝成 Node 節點插入主隊列末尾
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                // 拋出中斷異常
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

經過與前面的 acquire 方法對比能夠發現，兩者代碼幾乎同樣，區別在於 acquire 方法檢測到中斷（parkAndCheckInterrupt）時只是記錄了標誌位，並未響應；而此處直接拋出了異常。這也是兩者僅有的區別，此處再也不詳細分析。

2.2.3 以獨佔模式獲取資源（響應中斷，且有超時）

該操做與前者也是相似的，代碼以下：

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException 
    // 若被中斷，則響應
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}

doAcquireNanos:

static final long spinForTimeoutThreshold = 1000L;

private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    // 若超時時間小於等於 0，直接獲取失敗
    if (nanosTimeout <= 0L)
        return false;
    // 計算截止時間
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            nanosTimeout = deadline - System.nanoTime();
            // 已經超時了，獲取失敗
            if (nanosTimeout <= 0L)
                return false;
            // 若大於自旋時間，則線程休眠；不然自旋
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            // 若被中斷，則響應
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

這裏有個變量 spinForTimeoutThreshold，表示自旋時間，若大於該值則將線程休眠，不然繼續自旋。我的理解這裏增長該時間是爲了提升效率，即，只有在等待時間較長的時候才讓線程休眠。

該方法與 acquireInterruptibly 也是相似的，在前者的基礎上增長了 timeout，再也不詳細分析。

2.2.4 釋放資源

前面分析了三種獲取資源的方式，天然也有釋放資源。下面分析釋放資源的 release 操做：

public final boolean release(int arg) {
    // 嘗試釋放資源，若成功則返回 true
    if (tryRelease(arg)) {
        Node h = head;
        // 若頭節點不爲空，且等待狀態不爲 0（此時爲 SIGNAL）
        // 則喚醒其後繼節點
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

與 tryAcquire 方法相似，tryRelease 方法在 AQS 中也是拋出異常，一樣交由子類實現：

protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}

unparkSuccessor 的主要做用是喚醒 node 的後繼節點，代碼以下：

private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    // 後繼節點
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        // 若後繼節點是取消狀態，則從尾節點向前遍歷，找到 node 節點後面一個未取消狀態的節點
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    // 喚醒node節點的後繼節點
    if (s != null)
        LockSupport.unpark(s.thread);
}

若 node 節點的後繼節點是取消狀態（ws > 0），則從主隊列中取其後面一個非取消狀態的線程喚醒。

前面三個獲取資源的方法中，finally 代碼塊中都用到了 cancelAcquire 方法，都是獲取失敗時的操做，這裏也分析一下：

private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;
        
    node.thread = null;
    
    // Skip cancelled predecessors
    // 跳過取消狀態的前驅節點
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;
        
    // predNext is the apparent node to unsplice. CASes below will
    // fail if not, in which case, we lost race vs another cancel
    // or signal, so no further action is necessary.
    // 前驅節點的後繼節點引用
    Node predNext = pred.next;
    
    // Can use unconditional write instead of CAS here.
    // After this atomic step, other Nodes can skip past us.
    // Before, we are free of interference from other threads.
    // 將當前節點設置爲取消狀態
    node.waitStatus = Node.CANCELLED;
    
    // If we are the tail, remove ourselves.
    // 若該節點爲尾節點（後面沒其餘節點了），將 predNext 指向 null
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        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 {
            // 前驅節點爲頭節點，代表當前節點爲第一個，取消時喚醒它的下一個節點
            unparkSuccessor(node);
        }
        
        node.next = node; // help GC
    }
}

該方法的主要操做：

  1. 將 node 節點設置爲取消（CANCELLED）狀態；

  2. 找到它在隊列中非取消狀態的前驅節點 pred：

    1. 若 node 節點是尾節點，則前驅節點的後繼設爲空，
    2. 若 pred 不是頭節點，且狀態爲 SIGNAL，則後繼節點設爲 node 的後繼節點；
    3. 若 pred 是頭節點，則喚醒 node 的後繼節點。
PS: 該過程能夠跟雙鏈表刪除一個節點的過程進行對比分析。

3. 小結

本文分析了以獨佔模式獲取資源的三種方式，以及釋放資源的操做。分別爲：

  1. acquire: 獨佔模式獲取資源，忽略中斷；
  2. acquireInterruptibly: 獨佔模式獲取資源，響應中斷；
  3. tryAcquireNanos: 獨佔模式獲取資源，響應中斷，有超時；
  4. release: 釋放資源，喚醒主隊列中的下一個線程。

這幾個方法均可以類比 Lock 接口的相關方法定義。

相關閱讀：

JDK源碼分析-AbstractQueuedSynchronizer(1)

JDK源碼分析-Lock&Condition

image.png

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程