JDK源碼分析-AbstractQueuedSynchronizer(2)

概述node


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


Node 節點是對線程 Thread 類的封裝,所以兩種模式能夠理解以下:源碼分析

獨佔模式(exclusive):線程對資源的訪問是排他的,即某個時間只能一個線程單獨訪問資源;flex

共享模式(shared):與獨佔模式不一樣,多個線程能夠同時訪問資源。ui


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


獨佔模式atom


方法概述url


獨佔模式下的操做主要有如下幾個方法(可與前面分析的 Lock 接口的方法類比):spa


1. acquire(int arg)線程

以獨佔模式獲取資源,忽略中斷;能夠類比 Lock 接口的 lock 方法;


2. acquireInterruptibly(int arg)

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


3. tryAcquireNanos(int arg, long nanosTimeout)

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


4. release(int arg)

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


方法分析


1. 獨佔模式獲取資源(忽略中斷)


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

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>    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)) 方法的執行過程,爲了便於理解,可參考下面的流程圖:



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

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

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


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)時只是記錄了標誌位,並未響應;而此處直接拋出了異常。這也是兩者僅有的區別,此處再也不詳細分析。


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,再也不詳細分析。


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 {        




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:

    2.1 若 node 節點是尾節點,則前驅節點的後繼設爲空,

    2.2 若 pred 不是頭節點,且狀態爲 SIGNAL,則後繼節點設爲 node 的後繼節點;

    2.3 若 pred 是頭節點,則喚醒 node 的後繼節點。


PS: 該過程能夠跟雙鏈表刪除一個節點的過程進行對比分析。


小結


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


1. acquire: 獨佔模式獲取資源,忽略中斷;

2. acquireInterruptibly: 獨佔模式獲取資源,響應中斷;

3. tryAcquireNanos: 獨佔模式獲取資源,響應中斷,有超時;

4. release: 釋放資源,喚醒主隊列中的下一個線程。


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


相關閱讀:

JDK源碼分析-AbstractQueuedSynchronizer(1)

JDK源碼分析-Lock&Condition



Stay hungry, stay foolish.

相關文章
相關標籤/搜索