萬字長文，帶你深入理解 AQS 隊列同步器

 1Thread thread = new Thread(() -> {
 2    System.out.println("Thread start: " + Thread.currentThread().getName());
 3    LockSupport.park(); // 阻塞自己
 4    System.out.println("Thread end: " + Thread.currentThread().getName());
 5});
 6
 7thread.setName("A");
 8thread.start();
 9
10System.out.println("Main thread sleep 3 second: " + Thread.currentThread().getId());
11TimeUnit.SECONDS.sleep(3);
12LockSupport.unpark(thread); // 喚醒線程 A
13
14

1Thread start: A
2Main thread sleep 3 second: 1
3Thread end: A
4
5

1public static void park()
2public static void park(Object blocker)
3public static void parkNanos(long nanos)
4public static void parkNanos(Object blocker, long nanos)
5public static void parkUntil(long deadline)
6public static void parkUntil(Object blocker, long deadline)
7
8

 1public static void park(Object blocker) {
 2    // 獲取當前線程對象
 3    Thread t = Thread.currentThread();
 4    // 記錄當前線程阻塞等待的鎖對象（設置線程對象的 parkBlocker 爲參數指定的 blocker 對象）
 5    setBlocker(t, blocker);
 6    // 阻塞線程
 7    UNSAFE.park(false, 0L);
 8    // 線程恢復運行，清除 parkBlocker 參數記錄的鎖對象
 9    setBlocker(t, null);
10}
11
12

1public static void unpark(Thread thread) {
2    if (thread != null) {
3        UNSAFE.unpark(thread);
4    }
5}
6
7

 1static final class Node {
 2
 3    /** 模式定義 */
 4
 5    static final Node SHARED = new Node();
 6    static final Node EXCLUSIVE = null;
 7
 8    /** 線程狀態 */
 9
10    static final int CANCELLED = 1;
11    static final int SIGNAL = -1;
12    static final int CONDITION = -2;
13    static final int PROPAGATE = -3;
14
15    /** 線程等待狀態 */
16    volatile int waitStatus;
17
18    /** 前驅結點 */
19    volatile Node prev;
20    /** 後置結點 */
21    volatile Node next;
22
23    /** 持有的線程對象 */
24    volatile Thread thread;
25
26    /** 對於獨佔模式而言，指向下一個處於 CONDITION 等待狀態的結點；對於共享模式而言，則爲 SHARED 結點 */
27    Node nextWaiter;
28
29    // ... 省略方法定義
30}
31
32

1private transient volatile Node head;
2private transient volatile Node tail;
3
4

 1public class ConditionObject implements Condition, Serializable {
 2
 3        /** 指向條件隊列中的起始結點 */
 4        private transient Node firstWaiter;
 5        /** 指向條件隊列的末尾結點 */
 6        private transient Node lastWaiter;
 7
 8        // ... 省略方法定義
 9
10}
11
12

 1public interface Condition {
 2
 3    void await() throws InterruptedException;
 4    void awaitUninterruptibly();
 5    long awaitNanos(long nanosTimeout) throws InterruptedException;
 6    boolean await(long time, TimeUnit unit) throws InterruptedException;
 7    boolean awaitUntil(Date deadline) throws InterruptedException;
 8
 9    void signal();
10    void signalAll();
11}
12
13

 1public final void await() throws InterruptedException {
 2    if (Thread.interrupted()) {
 3        // 立即響應中斷
 4        throw new InterruptedException();
 5    }
 6    // 將當前線程添加到等待隊列末尾，等待狀態爲 CONDITION
 7    Node node = this.addConditionWaiter();
 8    // 釋放當前線程持有的資源
 9    int savedState = fullyRelease(node);
10    int interruptMode = 0;
11    while (!isOnSyncQueue(node)) { // 如果當前結點位於條件隊列中，則循環
12        // 阻塞當前線程
13        LockSupport.park(this);
14        // 如果線程在阻塞期間被中斷，則退出循環
15        if ((interruptMode = this.checkInterruptWhileWaiting(node)) != 0) {
16            break;
17        }
18    }
19    // 如果在同步隊列中等待期間被中斷，且之前的中斷狀態不爲 THROW_IE
20    if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
21        interruptMode = REINTERRUPT;
22    }
23    if (node.nextWaiter != null) {
24        // 清除條件隊列中所有狀態不爲 CONDITION 的結點
25        this.unlinkCancelledWaiters();
26    }
27    // 如果等待期間被中斷，則響應中斷
28    if (interruptMode != 0) {
29        this.reportInterruptAfterWait(interruptMode);
30    }
31}
32
33

 1private Node addConditionWaiter() {
 2    // 獲取條件隊列的末尾結點
 3    Node t = lastWaiter;
 4    // 如果末尾結點狀態不爲 CONDITION，表示對應的線程已經取消了等待，需要執行清理操作
 5    if (t != null && t.waitStatus != Node.CONDITION) {
 6        // 清除條件隊列中所有狀態不爲 CONDITION 的結點
 7        this.unlinkCancelledWaiters();
 8        t = lastWaiter;
 9    }
10    // 構建當前線程對應的 Node 結點，等待狀態爲 CONDITION，並添加到條件隊列末尾
11    Node node = new Node(Thread.currentThread(), Node.CONDITION);
12    if (t == null) {
13        firstWaiter = node;
14    } else {
15        t.nextWaiter = node;
16    }
17    lastWaiter = node;
18    return node;
19}
20
21

 1private void unlinkCancelledWaiters() {
 2    Node t = firstWaiter;
 3    Node trail = null; // 記錄上一個不被刪除的結點
 4    while (t != null) {
 5        Node next = t.nextWaiter;
 6        // 如果結點上的線程等待狀態不爲 CONDITION，則刪除對應結點
 7        if (t.waitStatus != Node.CONDITION) {
 8            t.nextWaiter = null;
 9            if (trail == null) {
10                firstWaiter = next;
11            } else {
12                trail.nextWaiter = next;
13            }
14            if (next == null) {
15                lastWaiter = trail;
16            }
17        } else {
18            trail = t;
19        }
20        t = next;
21    }
22}
23
24

 1final int fullyRelease(Node node) {
 2    boolean failed = true;
 3    try {
 4        // 獲取當前線程的同步狀態，可以理解爲持有的資源數量
 5        int savedState = this.getState();
 6        // 嘗試釋放當前線程持有的資源
 7        if (this.release(savedState)) {
 8            failed = false;
 9            return savedState;
10        } else {
11            // 釋放資源失敗
12            throw new IllegalMonitorStateException();
13        }
14    } finally {
15        if (failed) {
16            // 如果釋放資源失敗，則取消當前線程
17            node.waitStatus = Node.CANCELLED;
18        }
19    }
20}
21
22

 1final boolean isOnSyncQueue(Node node) {
 2    // 如果結點位於等待隊列，或是頭結點則返回 false
 3    if (node.waitStatus == Node.CONDITION || node.prev == null) {
 4        return false;
 5    }
 6    // If has successor, it must be on queue
 7    if (node.next != null) {
 8        return true;
 9    }
10
11    /*
12     * node.prev can be non-null, but not yet on queue because the CAS to place it on queue can fail.
13     * So we have to traverse from tail to make sure it actually made it. It will always be near the tail in calls to this method,
14     * and unless the CAS failed (which is unlikely), it will be there, so we hardly ever traverse much.
15     */
16
17    // 從後往前檢測目標結點是否位於同步隊列中
18    return this.findNodeFromTail(node);
19}
20
21

 1public final void awaitUninterruptibly() {
 2    // 將當前線程添加到等待隊列末尾，等待狀態爲 CONDITION
 3    Node node = this.addConditionWaiter();
 4    // 釋放當前線程持有的資源
 5    int savedState = fullyRelease(node);
 6    boolean interrupted = false;
 7    // 如果當前結點位於條件隊列中，則循環
 8    while (!isOnSyncQueue(node)) {
 9        // 阻塞當前線程
10        LockSupport.park(this);
11        if (Thread.interrupted()) {
12            // 標識線程等待期間被中斷，但不立即響應
13            interrupted = true;
14        }
15    }
16    // 自旋獲取資源，返回 true 則說明等待期間被中斷過
17    if (acquireQueued(node, savedState) || interrupted) {
18        // 響應中斷
19        selfInterrupt();
20    }
21}
22
23

 1public final void signal() {
 2    // 先檢測當前線程是否獲取到了鎖，否則不允許繼續執行
 3    if (!isHeldExclusively()) {
 4        throw new IllegalMonitorStateException();
 5    }
 6    // 獲取條件隊列頭結點，即等待時間最長的結點
 7    Node first = firstWaiter;
 8    if (first != null) {
 9        // 將頭結點從條件隊列轉移到同步隊列，參與競爭資源
10        this.doSignal(first);
11    }
12}
13
14

 1private void doSignal(Node first) {
 2    // 從前往後遍歷，直到遇到第一個不爲 null 的結點，並將其從條件隊列轉移到同步隊列
 3    do {
 4        if ((firstWaiter = first.nextWaiter) == null) {
 5            lastWaiter = null;
 6        }
 7        first.nextWaiter = null;
 8    } while (!transferForSignal(first) && (first = firstWaiter) != null);
 9}
10
11// AbstractQueuedSynchronizer#transferForSignal
12final boolean transferForSignal(Node node) {
13    // 更新當前結點的等待狀態：CONDITION -> 0
14    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
15        // 更新失敗，說明對應的結點上的線程已經被取消
16        return false;
17    }
18
19    /*
20     * Splice onto queue and try to set waitStatus of predecessor to indicate that thread is (probably) waiting.
21     * If cancelled or attempt to set waitStatus fails, wake up to resync (in which case the waitStatus can be transiently and harmlessly wrong).
22     */
23
24    // 將結點添加到同步隊列末端，並返回該結點的前驅結點
25    Node p = this.enq(node);
26    int ws = p.waitStatus;
27    // 如果前驅結點被取消，或者設置前驅結點的狀態爲 SIGNAL 失敗，則喚醒當前結點上的線程
28    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) {
29        LockSupport.unpark(node.thread);
30    }
31    return true;
32}
33
34

 1public final void signalAll() {
 2    if (!isHeldExclusively()) {
 3        // 先檢測當前線程是否獲取到了鎖，否則不允許繼續執行
 4        throw new IllegalMonitorStateException();
 5    }
 6    // 獲取條件隊列頭結點
 7    Node first = firstWaiter;
 8    if (first != null) {
 9        // 將所有結點從條件隊列轉移到同步隊列，參與競爭資源
10        this.doSignalAll(first);
11    }
12}
13
14private void doSignalAll(Node first) {
15    lastWaiter = firstWaiter = null;
16    do {
17        Node next = first.nextWaiter;
18        first.nextWaiter = null;
19        transferForSignal(first);
20        first = next;
21    } while (first != null);
22}
23
24

1private transient Thread exclusiveOwnerThread;
2
3private transient volatile Node head;
4private transient volatile Node tail;
5
6private volatile int state;
7
8

1protected boolean tryAcquire(int arg)
2protected boolean tryRelease(int arg)
3protected int tryAcquireShared(int arg)
4protected boolean tryReleaseShared(int arg)
5protected boolean isHeldExclusively()
6
7

 1public final void acquire(int arg) {
 2    if (!this.tryAcquire(arg) // 嘗試獲取資源
 3            // 如果獲取資源失敗，則將當前線程加入到同步隊列的末端（獨佔模式），並基於自旋機制等待獲取資源
 4            && this.acquireQueued(this.addWaiter(Node.EXCLUSIVE), arg)) {
 5        // 等待獲取資源期間曾被中斷過，在獲取資源成功之後再響應中斷
 6        selfInterrupt();
 7    }
 8}
 9
10

 1private Node addWaiter(Node mode) {
 2    // 爲當前線程創建結點對象
 3    Node node = new Node(Thread.currentThread(), mode);
 4    // 基於 CAS 機制嘗試快速添加結點到同步隊列末端
 5    Node pred = tail;
 6    if (pred != null) {
 7        node.prev = pred;
 8        if (this.compareAndSetTail(pred, node)) {
 9            pred.next = node;
10            return node;
11        }
12    }
13    // 快速添加失敗，繼續嘗試將該結點添加到同步隊列末端，如果同步隊列未被初始化則執行初始化
14    this.enq(node);
15    // 返回當前線程對應的結點對象
16    return node;
17}
18
19

 1private Node enq(final Node node) {
 2    for (; ; ) {
 3        // 獲取同步隊列末尾結點
 4        Node t = tail;
 5        // 如果結點不存在，則初始化
 6        if (t == null) { // Must initialize
 7            if (this.compareAndSetHead(new Node())) {
 8                tail = head;
 9            }
10        } else {
11            // 往末尾追加
12            node.prev = t;
13            if (this.compareAndSetTail(t, node)) {
14                t.next = node;
15                return t;
16            }
17        }
18    }
19}
20
21

 1final boolean acquireQueued(final Node node, int arg) {
 2    boolean failed = true;
 3    try {
 4        boolean interrupted = false; // 標記自旋過程中是否被中斷
 5        // 基於自旋機制等待獲取資源
 6        for (; ; ) {
 7            // 獲取前驅結點
 8            final Node p = node.predecessor();
 9            // 如果前驅結點爲頭結點，說明當前結點是排在同步隊列最前面，可以嘗試獲取資源
10            if (p == head && this.tryAcquire(arg)) {
11                // 獲取資源成功，更新頭結點
12                this.setHead(node); // 頭結點一般記錄持有資源的線程結點
13                p.next = null; // help GC
14                failed = false;
15                return interrupted; // 自旋過程中是否被中斷
16            }
17            // 如果還未輪到當前結點，或者獲取資源失敗
18            if (shouldParkAfterFailedAcquire(p, node) // 判斷是否需要阻塞當前線程
19                    && this.parkAndCheckInterrupt()) { // 如果需要，則進入阻塞狀態，並在甦醒時檢查中斷狀態
20                // 標識等待期間被中斷
21                interrupted = true;
22            }
23        }
24    } finally {
25        // 嘗試獲取資源失敗，說明執行異常，取消當前結點獲取資源的進程
26        if (failed) {
27            this.cancelAcquire(node);
28        }
29    }
30}
31
32

 1private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 2    // 獲取前驅結點狀態
 3    int ws = pred.waitStatus;
 4    if (ws == Node.SIGNAL) {
 5        // 前驅結點狀態爲 SIGNAL，說明當前結點需要被阻塞
 6        return true;
 7    }
 8    if (ws > 0) {
 9        // 前驅結點處於取消狀態，則一直往前尋找處於等待狀態的結點，並排在其後面
10        do {
11            node.prev = pred = pred.prev;
12        } while (pred.waitStatus > 0);
13        pred.next = node;
14    } else {
15        /*
16         * 前驅結點的狀態爲 0 或 PROPAGATE，但是當前結點需要一個被喚醒的信號，
17         * 所以基於 CAS 將前驅結點等待狀態設置爲 SIGNAL，在阻塞之前，調用者需要重試以再次確認不能獲取到資源。
18         */
19        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
20    }
21    return false;
22}
23
24

 1public final boolean release(int arg) {
 2    // 嘗試釋放資源
 3    if (this.tryRelease(arg)) {
 4        Node h = head;
 5        // 如果釋放資源成功，則嘗試喚醒後繼結點
 6        if (h != null && h.waitStatus != 0) {
 7            this.unparkSuccessor(h);
 8        }
 9        return true;
10    }
11    return false;
12}
13
14

 1private void unparkSuccessor(Node node) {
 2    // 獲取當前結點狀態
 3    int ws = node.waitStatus;
 4    if (ws < 0) {
 5        // 如果當前結點未被取消，則基於 CAS 更新結點等待狀態爲 0
 6        compareAndSetWaitStatus(node, ws, 0);
 7    }
 8
 9    /*
10     * Thread to unpark is held in successor, which is normally just the next node.
11     * But if cancelled or apparently null, traverse backwards from tail to find the actual non-cancelled successor.
12     */
13    Node s = node.next; // 獲取後繼結點
14    // 如果後繼結點爲 null，或者被取消
15    if (s == null || s.waitStatus > 0) {
16        s = null;
17        // 從後往前尋找距離當前結點最近的一個未被取消的線程結點
18        for (Node t = tail; t != null && t != node; t = t.prev) {
19            if (t.waitStatus <= 0) {
20                s = t;
21            }
22        }
23    }
24    // 喚醒結點上的線程
25    if (s != null) {
26        LockSupport.unpark(s.thread);
27    }
28}
29
30

1node.prev = pred;
2if (this.compareAndSetTail(pred, node)) {
3    pred.next = node;
4    return node;
5}
6
7

 1public final void acquireShared(int arg) {
 2    // 返回負數表示獲取資源失敗
 3    if (this.tryAcquireShared(arg) < 0) {
 4        // 將當前線程添加到條件隊列，基於自旋等待獲取資源
 5        this.doAcquireShared(arg);
 6    }
 7}
 8
 9private void doAcquireShared(int arg) {
10    // 將當前線程加入條件隊列末端，並標記爲共享模式
11    final Node node = this.addWaiter(Node.SHARED);
12    boolean failed = true;
13    try {
14        boolean interrupted = false; // 標記自旋過程中是否被中斷
15        for (; ; ) {
16            // 獲取前驅結點
17            final Node p = node.predecessor();
18            // 如果前驅結點爲頭結點，說明當前結點是排在同步隊列最前面，可以嘗試獲取資源
19            if (p == head) {
20                // 嘗試獲取資源
21                int r = this.tryAcquireShared(arg);
22                if (r >= 0) {
23                    // 獲取資源成功，設置自己爲頭結點，並嘗試喚醒後繼結點
24                    this.setHeadAndPropagate(node, r);
25                    p.next = null; // help GC
26                    if (interrupted) {
27                        selfInterrupt();
28                    }
29                    failed = false;
30                    return;
31                }
32            }
33            // 如果還未輪到當前結點，或者獲取資源失敗
34            if (shouldParkAfterFailedAcquire(p, node) // 判斷是否需要阻塞當前線程
35                    && this.parkAndCheckInterrupt()) { // 如果需要，則進入阻塞狀態，並在甦醒時檢查中斷狀態
36                // 標識等待期間被中斷
37                interrupted = true;
38            }
39        }
40    } finally {
41        // 嘗試獲取資源失敗，說明執行異常，取消當前結點獲取資源的進程
42        if (failed) {
43            this.cancelAcquire(node);
44        }
45    }
46}
47
48

 1private void setHeadAndPropagate(Node node, int propagate) {
 2    Node h = head; // 記錄之前的頭結點
 3    this.setHead(node); // 頭結點一般記錄持有資源的線程結點
 4    /*
 5     * 如果滿足以下條件，嘗試喚醒後繼結點：
 6     *
 7     * 1. 存在剩餘可用的資源；
 8     * 2. 後繼結點處於等待狀態，或後繼結點爲空
 9     *
10     * Try to signal next queued node if:
11     *   Propagation was indicated by caller,
12     *   or was recorded (as h.waitStatus either before or after setHead) by a previous operation
13     *   (note: this uses sign-check of waitStatus because PROPAGATE status may transition to SIGNAL.)
14     * and
15     *   The next node is waiting in shared mode,
16     *   or we don't know, because it appears null
17     *
18     * The conservatism in both of these checks may cause unnecessary wake-ups,
19     * but only when there are multiple racing acquires/releases, so most need signals now or soon anyway.
20     */
21    if (propagate > 0 // 存在剩餘可用的資源
22            || h == null || h.waitStatus < 0 // 此時 h 是之前的頭結點
23            || (h = head) == null || h.waitStatus < 0) { // 此時 h 已經更新爲當前頭結點
24        Node s = node.next;
25        // 如果後繼結點以共享模式在等待，或者後繼結點未知，則嘗試喚醒後繼結點
26        if (s == null || s.isShared()) {
27            this.doReleaseShared();
28        }
29    }
30}
31
32

 1public final boolean releaseShared(int arg) {
 2    // 嘗試釋放資源
 3    if (this.tryReleaseShared(arg)) {
 4        // 釋放資源成功，喚醒後繼結點
 5        this.doReleaseShared();
 6        return true;
 7    }
 8    return false;
 9}
10
11private void doReleaseShared() {
12    /*
13     * Ensure that a release propagates, even if there are other in-progress acquires/releases.
14     * This proceeds in the usual way of trying to unparkSuccessor of head if it needs signal.
15     * But if it does not, status is set to PROPAGATE to ensure that upon release, propagation continues.
16     * Additionally, we must loop in case a new node is added while we are doing this.
17     * Also, unlike other uses of unparkSuccessor, we need to know if CAS to reset status fails, if so rechecking.
18     */
19    for (; ; ) {
20        Node h = head;
21        if (h != null && h != tail) {
22            int ws = h.waitStatus;
23            // 如果頭結點狀態爲 SIGNAL，則在喚醒後繼結點之前嘗試清除當前結點的狀態
24            if (ws == Node.SIGNAL) {
25                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
26                    // loop to recheck cases
27                    continue;
28                }
29                // 喚醒後繼結點
30                this.unparkSuccessor(h);
31            }
32            /*
33             * 如果後繼結點暫時不需要被喚醒，則基於 CAS 嘗試將目標結點的 waitStatus 由 0 修改爲 PROPAGATE，
34             * 以保證後續由喚醒通知到來時，能夠將通知傳遞下去
35             */
36            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
37                // loop on failed CAS
38                continue;
39            }
40        }
41        // 如果頭結點未變更，則說明期間持有鎖的線程未發生變化，能夠走到這一步說明前面的操作已經成功完成
42        if (h == head) {
43            break;
44        }
45        // 如果頭結點發生變更，則說明期間持有鎖的線程發生了變化，需要重試以保證喚醒動作的成功執行
46    }
47}
48
49