AbstractQueuedSynchronizer 隊列同步器源碼分析

AbstractQueuedSynchronizer 隊列同步器（AQS）隊列同步器 (AQS), 是用來構建鎖或其他同步組件的基礎框架，它通過使用 int 變數表示同步狀態，通過內置的 FIFO 的隊列完成資源獲取的排隊工作。（摘自《Java併發編程的藝術》）我們知道獲取同步狀態有獨占和共用兩 ...

AbstractQueuedSynchronizer 隊列同步器（AQS）

隊列同步器 (AQS), 是用來構建鎖或其他同步組件的基礎框架，它通過使用 int 變數表示同步狀態，通過內置的 FIFO 的隊列完成資源獲取的排隊工作。（摘自《Java併發編程的藝術》）

我們知道獲取同步狀態有獨占和共用兩種模式，本文先針對獨占模式進行分析。

變數定義

private transient volatile Node head;

head 同步隊列頭節點

private transient volatile Node tail;

tail 同步隊列尾節點

private volatile int state;

state 同步狀態值

Node - 同步隊列節點定義

volatile int waitStatus;

waitStatus 節點的等待狀態，可取值如下 :

0 : 初始狀態
-1 : SIGNAL 處於該狀態的節點，說明其後置節點處於等待狀態；若當前節點釋放了鎖可喚醒後置節點
-2 : CONDITION 該狀態與 Condition 操作有關後續在說明
-3 : PROPAGATE 該狀態與共用式獲取同步狀態操作有關後續在說明
1 : CANCELLED 處於該狀態的節點會取消等待，從隊列中移除

volatile Node prev;

prev 指向當前節點的前置節點

volatile Node next;

next 指向當前節點的後置節點

volatile Thread thread;

thread 節點對應的線程也是指當前獲取鎖失敗的線程

Node nextWaiter;

acquire()

獨占模式下獲取同步狀態，既是當前只允許一個線程獲取到同步狀態

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

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

tryAcquire 嘗試獲取同步狀態，具體如何判定獲取到同步狀態由子類實現
當獲取同步狀態失敗時，執行 addWaiter 創建獨占模式下的 Node 並將其添加到同步隊列尾部
加入同步隊列之後，再次嘗試獲取同步狀態，當達到某種條件的時候將當前線程掛起等待喚醒

下麵具體看下各個階段如何實現：

private Node addWaiter(Node mode) {
    // 綁定當前線程 創建 Node 節點
    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 的前置節點指向隊列尾部
        node.prev = pred;
        // 將同步隊列的 tail 移動指向 node
        if (compareAndSetTail(pred, node)) {
            // 將原同步隊列的尾部後置節點指向 node
            pred.next = node;
            return node;
        }
    }
    // tail 為空說明同步隊列還未初始化
    // 此時調用 enq 完成隊列的初始化及 node 入隊
    enq(node);
    return node;
}

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

從代碼中可以看出通過 CAS 操作保證節點入隊的有序安全，其入隊過程中如下圖所示：

AQS節點入隊過程

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 
        for (;;) {
            // 獲取當前節點的前置節點
            final Node p = node.predecessor();
            // 判斷前置節點是否為 head 頭節點
            // 若前置節點為 head 節點，則再次嘗試獲取同步狀態
            if (p == head && tryAcquire(arg)) {
                // 若獲取同步狀態成功
                // 則將隊列的 head 移動指向當前節點
                setHead(node);
                // 將原頭部節點的 next 指向為空，便於對象回收
                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.
         */
        // 若前置節點狀態為 -1 ，則說明後置節點 node 可以安全掛起了
        return true;
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            // ws > 0 說明前置節點狀態為 CANCELLED , 也就是說前置節點為無效節點
            // 此時從前置節點開始向隊列頭節點方向尋找有效的前置節點
            // 此操作也即是將 CANCELLED 節點從隊列中移除
            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.
         */
        // 若前置節點狀態為初始狀態 則將其狀態設為 -1
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

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

從 acquireQueued 的實現可以看出，節點在入隊後會採用輪詢的方式（自旋）重覆執行以下過程：

判斷前置節點是否為 head，若為 head 節點則嘗試獲取同步狀態；若獲取同步狀態成功則移動 head 指向當前節點並退出迴圈
若前置節點非 head 節點或者獲取同步狀態失敗，則將前置節點狀態修改為 -1，並掛起當前線程，等待被喚醒重覆執行以上過程

如下圖所示：

AQS-節點自旋活動圖

接下來我們看看同步狀態釋放的實現。

release

釋放同步狀態

public final boolean release(int arg) {
    // 嘗試釋放同步狀態
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            // 喚醒後置節點
            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 節點狀態改為 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);
}

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

調用 tryRelease 嘗試釋放同步狀態，同樣其具體的實現由子類控制
成功釋放同步狀態後，將 head 節點狀態改為 0
喚醒後置節點上阻塞的線程

如下圖所示（紅色曲線表示節點自旋過程） :

AQS-釋放鎖

acquireInterruptibly()

獨占模式下獲取同步狀態，不同於 acquire 方法，該方法對中斷操作敏感；也就是說當前線程在獲取同步狀態的過程中，若被中斷則會拋出中斷異常

public final void acquireInterruptibly(int arg)
            throws InterruptedException {
    if (Thread.interrupted())
        // 檢查線程是否被中斷
        // 中斷則拋出中斷異常由調用方處理
        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)
            // 拋出中斷異常後最終執行 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.
    // 若當前節點為 tail 節點，則將 tail 移動指向 node 的前置節點
    if (node == tail && compareAndSetTail(node, pred)) {
        // 同時將node 前置節點的 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) {
            // 當前節點位於隊列中部    
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                // 將前置節點的 next 指向 node 的後置節點
                compareAndSetNext(pred, predNext, next);
        } else {
            // 若 node 的前置節點為 head 節點則喚醒 node 節點的後置節點
            unparkSuccessor(node);
        }

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

從 acquireInterruptibly 的實現可以看出，若線程在獲取同步狀態的過程中出現中斷操作，則會將當前線程對應的同步隊列等待節點從隊列中移除並喚醒可獲取同步狀態的線程。

tryAcquireNanos()

獨占模式超時獲取同步狀態，該操作與acquireInterruptibly一樣對中斷操作敏感，不同在於超過等待時間若未獲取到同步狀態將會返回

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;
    // 計算等待到期時間
    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);
    }
}

節點的狀態

同步隊列中的節點在自旋獲取同步狀態的過程中，會將前置節點的狀態由 0 初始狀態改為 -1 SIGNAL, 若是中斷敏感的操作則會將狀態由 0 改為 1

同步隊列中的節點在釋放同步狀態的過程中會將同步隊列的 head 節點的狀態改為 0，也即是由 -1 變為 0；

小結

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

在獲取同步狀態時，AQS 內部維護了一個同步隊列，獲取狀態失敗的線程會被構造一個節點加入到隊列中併進行一系列自旋操作
在釋放同步狀態時，喚醒 head 的後置節點去獲取同步狀態