AbstractQueuedSynchronizer 隊(duì)列同步器（AQS）

隊(duì)列同步器 (AQS), 是用來構(gòu)建鎖或其他同步組件的基礎(chǔ)框架侦锯，它通過使用 int 變量表示同步狀態(tài)吱抚，通過內(nèi)置的 FIFO 的隊(duì)列完成資源獲取的排隊(duì)工作剩膘。（摘自《Java并發(fā)編程的藝術(shù)》）

我們知道獲取同步狀態(tài)有獨(dú)占和共享兩種模式仅叫，本文先針對獨(dú)占模式進(jìn)行分析庞萍。

變量定義

private transient volatile Node head;

head 同步隊(duì)列頭節(jié)點(diǎn)

private transient volatile Node tail;

tail 同步隊(duì)列尾節(jié)點(diǎn)

private volatile int state;

state 同步狀態(tài)值

Node - 同步隊(duì)列節(jié)點(diǎn)定義

volatile int waitStatus;

waitStatus 節(jié)點(diǎn)的等待狀態(tài)，可取值如下 :

• 0 : 初始狀態(tài)
• -1 : SIGNAL 處于該狀態(tài)的節(jié)點(diǎn)痛倚，說明其后置節(jié)點(diǎn)處于等待狀態(tài)丰刊； 若當(dāng)前節(jié)點(diǎn)釋放了鎖可喚醒后置節(jié)點(diǎn)
• -2 : CONDITION 該狀態(tài)與 Condition 操作有關(guān)后續(xù)在說明
• -3 : PROPAGATE 該狀態(tài)與共享式獲取同步狀態(tài)操作有關(guān)后續(xù)在說明
• 1 : CANCELLED 處于該狀態(tài)的節(jié)點(diǎn)會取消等待，從隊(duì)列中移除

volatile Node prev;

prev 指向當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)

volatile Node next;

next 指向當(dāng)前節(jié)點(diǎn)的后置節(jié)點(diǎn)

volatile Thread thread;

thread 節(jié)點(diǎn)對應(yīng)的線程也是指當(dāng)前獲取鎖失敗的線程

Node nextWaiter;

acquire()

獨(dú)占模式下獲取同步狀態(tài)瞒斩， 既是當(dāng)前只允許一個線程獲取到同步狀態(tài)

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

從 acquire 方法中我們可以大概猜測下破婆，獲取鎖的過程如下：

• tryAcquire 嘗試獲取同步狀態(tài)， 具體如何判定獲取到同步狀態(tài)由子類實(shí)現(xiàn)
• 當(dāng)獲取同步狀態(tài)失敗時胸囱，執(zhí)行 addWaiter 創(chuàng)建獨(dú)占模式下的 Node 并將其添加到同步隊(duì)列尾部
• 加入同步隊(duì)列之后荠割，再次嘗試獲取同步狀態(tài)，當(dāng)達(dá)到某種條件的時候?qū)?dāng)前線程掛起等待喚醒

下面具體看下各個階段如何實(shí)現(xiàn)：

private Node addWaiter(Node mode) {
    // 綁定當(dāng)前線程 創(chuàng)建 Node 節(jié)點(diǎn)
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    // 判斷同步隊(duì)列尾節(jié)點(diǎn)是否為空
    if (pred != null) {
        // node 的前置節(jié)點(diǎn)指向隊(duì)列尾部
        node.prev = pred;
        // 將同步隊(duì)列的 tail 移動指向 node
        if (compareAndSetTail(pred, node)) {
            // 將原同步隊(duì)列的尾部后置節(jié)點(diǎn)指向 node
            pred.next = node;
            return node;
        }
    }
    // tail 為空說明同步隊(duì)列還未初始化
    // 此時調(diào)用 enq 完成隊(duì)列的初始化及 node 入隊(duì)
    enq(node);
    return node;
}

private Node enq(final Node node) {
    // 輪詢的方式執(zhí)行
    // 成功入隊(duì)后退出
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            // 創(chuàng)建 Node, 并將 head 指向該節(jié)點(diǎn)
            // 同時將 tail 指向該節(jié)點(diǎn)
            // 完成隊(duì)列的初始化
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // node 的前置節(jié)點(diǎn)指向隊(duì)列尾部
            node.prev = t;
            // 將同步隊(duì)列的 tail 移動指向 node
            if (compareAndSetTail(t, node)) {
                // 將原同步隊(duì)列的尾部后置節(jié)點(diǎn)指向 node
                t.next = node;
                return t;
            }
        }
    }
}

從代碼中可以看出通過 CAS 操作保證節(jié)點(diǎn)入隊(duì)的有序安全旺矾，其入隊(duì)過程中如下圖所示：

AQS節(jié)點(diǎn)入隊(duì)過程

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 
        for (;;) {
            // 獲取當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)
            final Node p = node.predecessor();
            // 判斷前置節(jié)點(diǎn)是否為 head 頭節(jié)點(diǎn)
            // 若前置節(jié)點(diǎn)為 head 節(jié)點(diǎn)蔑鹦，則再次嘗試獲取同步狀態(tài)
            if (p == head && tryAcquire(arg)) {
                // 若獲取同步狀態(tài)成功
                // 則將隊(duì)列的 head 移動指向當(dāng)前節(jié)點(diǎn)
                setHead(node);
                // 將原頭部節(jié)點(diǎn)的 next 指向?yàn)榭眨阌趯ο蠡厥?                p.next = null; // help GC
                failed = false;
                // 退出輪詢過程
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

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.
         */
        // 若前置節(jié)點(diǎn)狀態(tài)為 -1 箕宙，則說明后置節(jié)點(diǎn) node 可以安全掛起了
        return true;
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            // ws > 0 說明前置節(jié)點(diǎn)狀態(tài)為 CANCELLED , 也就是說前置節(jié)點(diǎn)為無效節(jié)點(diǎn)
            // 此時從前置節(jié)點(diǎn)開始向隊(duì)列頭節(jié)點(diǎn)方向?qū)ふ矣行У那爸霉?jié)點(diǎn)
            // 此操作也即是將 CANCELLED 節(jié)點(diǎn)從隊(duì)列中移除
            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.
         */
        // 若前置節(jié)點(diǎn)狀態(tài)為初始狀態(tài) 則將其狀態(tài)設(shè)為 -1
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

private final boolean parkAndCheckInterrupt() {
    // 將當(dāng)前線程掛起
    LockSupport.park(this);
    // 被喚醒后檢查當(dāng)前線程是否被掛起
    return Thread.interrupted();
}

從 acquireQueued 的實(shí)現(xiàn)可以看出嚎朽，節(jié)點(diǎn)在入隊(duì)后會采用輪詢的方式（自旋）重復(fù)執(zhí)行以下過程：

• 判斷前置節(jié)點(diǎn)是否為 head， 若為 head 節(jié)點(diǎn)則嘗試獲取同步狀態(tài)柬帕； 若獲取同步狀態(tài)成功則移動 head 指向當(dāng)前節(jié)點(diǎn)并退出循環(huán)
• 若前置節(jié)點(diǎn)非 head 節(jié)點(diǎn)或者獲取同步狀態(tài)失敗哟忍，則將前置節(jié)點(diǎn)狀態(tài)修改為 -1， 并掛起當(dāng)前線程陷寝，等待被喚醒重復(fù)執(zhí)行以上過程

如下圖所示：

AQS-節(jié)點(diǎn)自旋活動圖

接下來我們看看同步狀態(tài)釋放的實(shí)現(xiàn)锅很。

release

釋放同步狀態(tài)

public final boolean release(int arg) {
    // 嘗試釋放同步狀態(tài)
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            // 喚醒后置節(jié)點(diǎn)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

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)
        // 將 head 節(jié)點(diǎn)狀態(tài)改為 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.
     */
    // 獲取后置節(jié)點(diǎn)
    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)
        // 喚醒后置節(jié)點(diǎn)上所阻塞的線程
        LockSupport.unpark(s.thread);
}

從上述代碼，我們可以明白釋放同步狀態(tài)的過程如下：

• 調(diào)用 tryRelease 嘗試釋放同步狀態(tài)凤跑，同樣其具體的實(shí)現(xiàn)由子類控制
• 成功釋放同步狀態(tài)后爆安，將 head 節(jié)點(diǎn)狀態(tài)改為 0
• 喚醒后置節(jié)點(diǎn)上阻塞的線程

如下圖所示（紅色曲線表示節(jié)點(diǎn)自旋過程） :

AQS-釋放鎖

acquireInterruptibly()

獨(dú)占模式下獲取同步狀態(tài)，不同于 acquire 方法仔引，該方法對中斷操作敏感扔仓； 也就是說當(dāng)前線程在獲取同步狀態(tài)的過程中褐奥，若被中斷則會拋出中斷異常

public final void acquireInterruptibly(int arg)
            throws InterruptedException {
    if (Thread.interrupted())
        // 檢查線程是否被中斷
        // 中斷則拋出中斷異常由調(diào)用方處理
        throw new InterruptedException();
    if (!tryAcquire(arg))
        doAcquireInterruptibly(arg);
}

private void doAcquireInterruptibly(int arg)
        throws InterruptedException {
    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())
                // 不同于 acquire 的操作，此處在喚醒后檢查是否中斷翘簇，若被中斷直接拋出中斷異常
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            // 拋出中斷異常后最終執(zhí)行 cancelAcquire
            cancelAcquire(node);
    }
}

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.
    // 若當(dāng)前節(jié)點(diǎn)為 tail 節(jié)點(diǎn)撬码，則將 tail 移動指向 node 的前置節(jié)點(diǎn)
    if (node == tail && compareAndSetTail(node, pred)) {
        // 同時將node 前置節(jié)點(diǎn)的 next 指向 null
        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) {
            // 當(dāng)前節(jié)點(diǎn)位于隊(duì)列中部    
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                // 將前置節(jié)點(diǎn)的 next 指向 node 的后置節(jié)點(diǎn)
                compareAndSetNext(pred, predNext, next);
        } else {
            // 若 node 的前置節(jié)點(diǎn)為 head 節(jié)點(diǎn)則喚醒 node 節(jié)點(diǎn)的后置節(jié)點(diǎn)
            unparkSuccessor(node);
        }

        node.next = node; // help GC
    }
}

從 acquireInterruptibly 的實(shí)現(xiàn)可以看出，若線程在獲取同步狀態(tài)的過程中出現(xiàn)中斷操作版保，則會將當(dāng)前線程對應(yīng)的同步隊(duì)列等待節(jié)點(diǎn)從隊(duì)列中移除并喚醒可獲取同步狀態(tài)的線程呜笑。

tryAcquireNanos()

獨(dú)占模式超時獲取同步狀態(tài)，該操作與acquireInterruptibly一樣對中斷操作敏感彻犁，不同在于超過等待時間若未獲取到同步狀態(tài)將會返回

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}

private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    // 計(jì)算等待到期時間
    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)
                // 按指定時間掛起s
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

節(jié)點(diǎn)的狀態(tài)

同步隊(duì)列中的節(jié)點(diǎn)在自旋獲取同步狀態(tài)的過程中叫胁，會將前置節(jié)點(diǎn)的狀態(tài)由 0 初始狀態(tài)改為 -1 SIGNAL, 若是中斷敏感的操作則會將狀態(tài)由 0 改為 1

同步隊(duì)列中的節(jié)點(diǎn)在釋放同步狀態(tài)的過程中會將同步隊(duì)列的 head 節(jié)點(diǎn)的狀態(tài)改為 0， 也即是由 -1 變?yōu)?0袖裕；

小結(jié)

本文主要分析了獨(dú)占模式獲取同步狀態(tài)的操作，其大概流程如下：

• 在獲取同步狀態(tài)時溉瓶，AQS 內(nèi)部維護(hù)了一個同步隊(duì)列急鳄，獲取狀態(tài)失敗的線程會被構(gòu)造一個節(jié)點(diǎn)加入到隊(duì)列中并進(jìn)行一系列自旋操作
• 在釋放同步狀態(tài)時，喚醒 head 的后置節(jié)點(diǎn)去獲取同步狀態(tài)