AbstractQueuedSynchronizer 的示例分析

這篇文章給大家介紹AbstractQueuedSynchronizer 的示例分析，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

AbstractQueuedSynchronizer

AQS是一個(gè)內(nèi)部維護(hù)了先進(jìn)先出隊(duì)列+標(biāo)識(shí)(數(shù)字)的模型，標(biāo)識(shí)使用CAS模式修改，作為多線程工具的基礎(chǔ)組件

屬性

state為volatile的int，不同的業(yè)務(wù)場(chǎng)景按需實(shí)現(xiàn)

獨(dú)占模式：

ReentrantLock.Sync中state為0表示未鎖定>0表示被幾個(gè)線程持有

ThreadPoolExecutor.Worker中state為0表示未執(zhí)行

共享模式：

CountDownLatch.Sync中state為初始化時(shí)指定,表示有多少個(gè)線程可持有,

Semaphore.Sync中state與CountDownLatch相同

混合模式:

ReentrantReadWriteLock.Sync中state為共享鎖+獨(dú)占鎖組合 通過(guò)位運(yùn)算16位來(lái)分割,最大的讀鎖寫鎖個(gè)數(shù)為65535

關(guān)鍵方法

獨(dú)占模式

持有資源：acquire

acquire: 獨(dú)占模式獲取,獨(dú)占模式即只有一個(gè)線程可更新state.忽略中斷標(biāo)識(shí)，在獲取之后響應(yīng)中斷。

acquireInterruptibly:獨(dú)占模式獲取,線程標(biāo)識(shí)為中斷則拋出異常

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

tryAcquire:子類按需實(shí)現(xiàn),使用獨(dú)占模式更新state,增加state，成功返回true失敗返回false

中斷后不會(huì)正確響應(yīng)park,所以需要重置線程中斷標(biāo)識(shí),并在unpark之后進(jìn)行中斷補(bǔ)償

釋放資源：release

release:以獨(dú)占模式釋放資源

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

tryRelease:子類按需實(shí)現(xiàn),使用獨(dú)占模式更新state，減少state,并處理對(duì)應(yīng)獨(dú)占線程

共享模式

持有資源:acquireShared

• acquireShared 共享模式獲取，忽略中斷線程，在獲取之后相應(yīng)中斷。

• acquireSharedInterruptibly 共享模式獲取，響應(yīng)中斷，線程中斷則拋出異常。

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

tryAcquireShared:子類按需實(shí)現(xiàn)，對(duì)返回值有如下要求：

負(fù)值:失敗。 零:共享模式下的獲取成功，但是沒有后續(xù)共享模式下的獲取可以成功。 正值: 如果共享模式下的獲取成功并且后續(xù)共享模式下的獲取也可能成功，則為正值，在這種情況下，后續(xù)的等待線程必須檢查可用性。 （對(duì)三個(gè)不同返回值的支持使該方法可以在僅有時(shí)進(jìn)行獲取的情況下使用。）成功后，就已經(jīng)獲取了此對(duì)象。

釋放資源:releaseShared

• releaseShared:以共享模式釋放資源

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

tryReleaseShared:子類按需實(shí)現(xiàn),使用共享模式更新state，減少state

其它關(guān)鍵方法

檢查并更新節(jié)點(diǎn)狀態(tài):shouldParkAfterFailedAcquire

在park線程之前的判斷,當(dāng)前置節(jié)點(diǎn)為取消時(shí)更新前置節(jié)點(diǎn)

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        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.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

喚醒后續(xù)節(jié)點(diǎn):unparkSuccessor

喚醒后續(xù)節(jié)點(diǎn),并清理取消的節(jié)點(diǎn),

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) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

共享模式設(shè)置隊(duì)列頭:setHeadAndPropagate

共享模式下多個(gè)線程同時(shí)持有資源,頭節(jié)點(diǎn)會(huì)頻繁變化需要及時(shí)釋放資源

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);
    /*
     * Try to signal next queued node if:
     *   Propagation was indicated by caller,
     *     or was recorded (as h.waitStatus either before
     *     or after setHead) by a previous operation
     *     (note: this uses sign-check of waitStatus because
     *      PROPAGATE status may transition to SIGNAL.)
     * and
     *   The next node is waiting in shared mode,
     *     or we don't know, because it appears null
     *
     * The conservatism in both of these checks may cause
     * unnecessary wake-ups, but only when there are multiple
     * racing acquires/releases, so most need signals now or soon
     * anyway.
     */
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

Node

關(guān)于AbstractQueuedSynchronizer 的示例分析就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到。