Java 同步鎖ReentrantLock與抽象同步佇列AQS

AbstractQueuedSynchronizer 抽象同步佇列，它是個模板類提供了許多以鎖相關的操作，常說的AQS指的就是它，AQS繼承了AbstractOwnableSynchronizer類，AOS用于保存執行緒物件，保存什么執行緒物件呢？保存鎖被獨占的執行緒物件，

抽象同步佇列AQS除了實作序列化標記介面，并沒有實作任何的同步介面，該類提供了許多同步狀態獲取和釋放的方法給自定義同步器使用，如ReentrantLock的內部類Sync，抽象同步佇列支持獨占式或共享式的的獲取同步狀態，方便實作不同型別的自定義同步器，一般方法名帶有Shared的為共享式，比如，嘗試以共享式的獲取鎖的方法int tryAcquireShared(int)，而獨占式獲取鎖方法為boolean tryAcquire(int)，

AQS是抽象同步佇列，其重點就是同步佇列及如何操作同步佇列，

同步佇列

雙向同步佇列，采用尾插法新增節點，從頭部的下一個節點獲取操作節點，節點自旋獲取同步鎖，實作FIFO（先進先出）原則，

理解節點中的屬性值作用

• prev：前驅節點；即當前節點的前一個節點，之所以叫前驅節點，是因為前一個節點在使用完鎖之后會解除后一個節點的阻塞狀態；

• next：后繼節點；即當前節點的后一個節點，之所以叫后繼節點，是因為“后繼有人”了，表示有“下一代”節點承接這個獨有的鎖??；

• nextWaiter：表示指向下一個Node.CONDITION狀態的節點（本文不講述Condition佇列，在此可以忽略它）；

• thread：節點物件中保存的執行緒物件，節點都是配角，執行緒才是主角；

• waitStatus：當前節點在佇列中的等待狀態；waitStatus = CANCELLED = 1，表示執行緒已經取消（該狀態下的節點為作廢節點，將從佇列中斷開）；

  • waitStatus = SIGNAL = -1，表示執行緒處于請求釋放的狀態，后繼執行緒需要阻塞等待（該狀態下的節點執行緒處于阻塞等待狀態或獲取鎖未釋放狀態）；

  • waitStatus = CONDITION = -2，表示執行緒正在等待；

  • waitStatus = PROPAGATE = -3，在共享情況下，表示下一個被請求的shared應該無條件傳播；

  • waitStatus = 0，表示節點初始化時的默認值（int型別成員變數的默認值），

注意1：節點物件中的prev、next和nextWaiter都是一個完整的Node節點物件，也就是說每個節點都保存了前后節點的物件，如果沒有則為null，

注意2：head節點是個虛節點（prev=null、thread=null），但head本身是一個實際存在的節點物件，起到標記佇列的開頭；尾節點tail節點的next=null，等待新節點插入，

頭節點head為什么是虛節點

重點：必須明確知道頭節點head為什么是虛節點!!!這很重要，

原因是，當前節點在獲取到鎖??之后，它這個執行緒物件就會被保存到AOS（AbstractOwnableSynchronizer）中的exclusiveOwnerThread 物件，（一開頭就提到過了，在這里再強調一次），所以在佇列中，頭節點是無需存盤Thread物件的了，那為什么設計成這樣呢？因為存在臨界情況就是只有一個執行緒獲取鎖資源時，無需初始化生成同步佇列，直接獲取同步鎖即可，只有存在鎖未釋放同時又進來了新的執行緒時，才會去初始化同步佇列，并為未釋放鎖的執行緒占個位置，這個位置就是頭節點head，表明前面還有個執行緒在使用資源，

// 初始化佇列的方法private Node enq(final Node node) {
    // 死回圈
    for (;;) {
        Node t = tail;
        // 沒頭沒尾時
        if (t == null) { // Must initialize
            // 生成頭節點（尾節點）
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 有頭有尾后，才把需要等待的執行緒節點加入佇列中
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

在獨占執行緒釋放鎖時，判斷head是否為null，即可知道同步佇列是否存在，如果同步佇列不存在，那么無需執行嘗試喚醒后繼節點那些操作了，

在同個時間節點中，單個執行緒只需要操作AOS的Thread物件和AQS的state狀態即可實作同步鎖和鎖的可重入性，

執行緒加入同步佇列的程序

在鎖被占用時，獲取鎖失敗后，當前執行緒被封裝成Node節點并加入到佇列尾部，

即tryAcquire(arg)回傳false時，

執行acquireQueued(addWaiter(Node.EXCLUSIVE), arg))操作，

而addWaiter(Node.EXCLUSIVE) 為新增節點到同步佇列，佇列未初始化時會執行enq完成初始化后再新增節點到佇列，

因此，新增節點分為首次生成同步佇列同時新增節點和在原有同步佇列中插入新增節點，

首次生成同步佇列新增節點：通過enq(final Node node)方法，先初始化頭節點（虛節點），再通過原子操作compareAndSetTail方法從佇列尾部插入新節點，

在原有同步佇列中新增節點：通過原子操作compareAndSetTail方法從佇列尾部插入新節點，

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

private Node addWaiter(Node mode) {
    // 使用當前執行緒生成新節點
    Node node = new Node(Thread.currentThread(), mode);
    // 獲取同步器的尾節點
    Node pred = tail;
    if (pred != null) {
        // 第一步：新節點的prev節點指向尾部節點（pred=tail）
        node.prev = pred;
        // 第二部：CAS比較尾節點，相等就讓tail=node
        if (compareAndSetTail(pred, node)) {
            // 第三步：pred=舊的tail，即舊的尾節點的next節點指向新節點
            pred.next = node;
            return node;
        }
    }
    // 當tail=null 時執行，邏輯相類似的；enq初始化的head節點為虛節點
    enq(node);
    return node;
}

// 將節點插入佇列，必要時初始化（即tail=null時，也即是同步佇列沒有節點時初始化），private Node enq(final Node node) {
    // 死回圈
    for (;;) {
        // 第一次進來tail=null，第二次進來tail=head=new Node()
        Node t = tail;
        if (t == null) {
            // 創建一個空節點作為head節點
            if (compareAndSetHead(new Node()))
                tail = head;
        } else { // 下面就是正常的尾插法新增節點
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

配合原始碼和動圖理解：新增佇列節點程序（三部曲）

使用尾插法新增同步佇列節點

• 第一步：新增節點的prev節點指向尾節點tail；

• 第二步：尾節點tail 和新節點做CAS操作，即compareAndSetTail(pred,node) ,即同步器的tail節點指向新節點；

• 第三步：舊的尾節點的next節點指向新節點（此時的新節點=尾節點tail）

最終結果圖

新增節點加入佇列之后，在同步佇列中執行緒怎么等待？執行緒怎么獲取鎖呢？

節點執行緒獲取鎖

看代碼前必須明確知道哪個節點是要獲取鎖的，頭節點為虛節點，標記佇列的開頭，真正要獲取鎖的是頭節點的后繼節點，

獲取鎖程序

鎖在釋放時呼叫的關鍵流程：

ReentrantLock#lock() -> Sync#lock() -> AQS#acquire(1) -> NonfairSync#tryAcquire(1) 【或FairSync#tryAcquire(1)】-> AQS#addWaiter(node) -> AQS#acquireQueued(node,1) -> AQS#selfInterrupt()

關鍵代碼??

// java.util.concurrent.locks.AbstractQueuedSynchronizer/**
 * 以獨占不可中斷模式獲取已在佇列中的執行緒，
 */final boolean acquireQueued(final Node node, int arg) {
    // 例外標志狀態
    boolean failed = true;
    try {
        // 是否發生過中斷的標志
        boolean interrupted = false;
        // 自旋鎖>>>死回圈：每個node都獨立執行著這個死回圈，直至執行緒被阻塞
        for (;;) {
            // 獲取前驅節點（當前節點的前一個節點）
            final Node p = node.predecessor();
            // 當前節點的前驅節點是否等于頭節點（虛節點），等于就會執行嘗試獲取鎖 tryAcquire(arg)
            if (p == head && tryAcquire(arg)) {
                setHead(node); // 設定當前節點為頭節點，在獲取鎖成功時，thread物件已經保存到AQS中的exclusiveOwnerThread了
                p.next = null; // 前驅節點的next指向null，斷開前驅節點（舊的頭節點）
                failed = false; // 只要當前節點node正常獲取到鎖，就不會執行finally的cancelAcquire(node)
                return interrupted;
            }
            // 前驅節點的waitStatus=-1時，當前node會被阻塞，防止無限回圈浪費資源
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        /* 正常情況：只有return時，才會執行finally代碼，而只要return，failed都等于false，
         * 所以，failed 是為了避免節點發生例外時，node沒有被釋放，
         * 比如：node.predecessor() 可能產生空指標例外，
         */
        if (failed)
            cancelAcquire(node);
    }
}


// java.util.concurrent.locks.AbstractQueuedSynchronizerprivate void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

為什么在嘗試獲取鎖前要判斷前驅節點是否為頭節點？

因為除了阻塞被釋放會讓死回圈繼續執行的情況外，還有中斷指令也會使執行緒從阻塞狀態中被釋放，所以存在任意節點提前重新執行“死回圈”嘗試獲取鎖的情況，如果不判斷獲取鎖節點的前驅節點是否為頭節點，那就會出現提前嘗試獲取鎖，從而破壞了同步佇列的先進先出（FIFO）原則，說白了，就是被插隊了，

當前節點不是頭節點時，執行以下代碼

// java.util.concurrent.locks.AbstractQueuedSynchronizer

/**
 * 檢查和更新獲取鎖失敗的節點的狀態，如果執行緒需要等待，則回傳true，使其執行阻塞操作，
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 獲取前驅節點的等待狀態
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        // 因為前驅節點處于請求釋放的狀態，所以當前節點需要阻塞等待，會回傳true，從而執行后續方法進入阻塞狀態
        return true;
    if (ws > 0) {
        // 前驅節點被標上取消標志了，需要跳過前驅節點并不斷重試
        do {
            // 回圈向前查找取消節點，把取消節點從佇列中移除
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * waitStatus必須為0或等于PROPAGATE=-3
         * 表示需要設定前驅節點等待狀態為SIGNAL，
         * 將會在外層回圈再次嘗試獲取鎖，如果再次獲取鎖失敗，那么就會阻塞當前執行緒
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

當shouldParkAfterFailedAcquire(p, node) 回傳true時，將會執行阻塞操作parkAndCheckInterrupt())，其通過執行緒阻塞工具類方法LockSupport.park(this) 來阻塞當前執行緒，

private final boolean parkAndCheckInterrupt() {
    // 阻塞當前執行緒執行緒調度
    LockSupport.park(this);
    // 清除當前執行緒的中斷狀態，并回傳上一次的中斷狀態
    return Thread.interrupted();
}

注意：LockSupport.park(this) 阻塞后，需要喚醒阻塞才會執行后續操作，可通過解除阻塞LockSupport.unpark(thread) 或 中斷thread.interrupt() 來喚醒阻塞，

只有shouldParkAfterFailedAcquire 和 parkAndCheckInterrupt都回傳true時，才會執行interrupted = true，即只有是中斷導致阻塞結束時，才回傳true，此時 selfInterrupt()重新執行一次中斷操作，

// java.util.concurrent.locks.AbstractQueuedSynchronizer
final boolean acquireQueued(final Node node, int arg) {
    ......
    if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    // 只有是中斷導致阻塞結束時，才回傳true
                    interrupted = true;
    ......
}

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // 中斷當前執行緒
        selfInterrupt();
}

static void selfInterrupt() {
    // 中斷當前執行緒
    Thread.currentThread().interrupt();
}

為什么需要再一次執行中斷呢？

因為存在中斷thread.interrupt() 喚醒自旋鎖阻塞的情況，而Thread.interrupted() 獲取中斷狀態并清除當前執行緒的中斷狀態，所以需要重新執行一次中斷操作selfInterrupt()，將中斷標志置為true，

這種獲取鎖的方式是非中斷鎖，就是無法通過中斷的方式結束鎖的獲取，區別于中斷鎖，所以該方式在獲取鎖的程序中，不會處理中斷，只是記錄中斷狀態，Thread.interrupted() 獲取中斷狀態后清除中斷狀態，所以需要重新設定中斷標志為true，

如果你想要處理中斷的情況，那我們可以在acquireQueued(addWaiter(Node.EXCLUSIVE), arg) 回傳true 的時候去處理，比如：拋出中斷例外，

如果你需要在執行緒發生中斷時結束獲取鎖，那么可以考慮使用lockInterruptibly()來獲取鎖，

兩種方式獲取鎖的區別

lock()方式獲取鎖：自旋鎖只會在正常獲取到鎖或發生例外時結束自旋鎖（死回圈），

void lockInterruptibly() 方式獲取鎖：會在發生中斷的情況下，拋出中斷例外throw new InterruptedException(); 從而跳出自旋鎖（死回圈），而呼叫lockInterruptibly() 的方法需要捕獲中斷例外，做一些例外處理，

獲取鎖的lock()和lockInterruptibly()的主要區別是：中斷是否會結束鎖的獲取，

看看原始碼怎么實作中斷結束鎖的獲取

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())
    // 只有是中斷導致阻塞結束時，才拋出中斷例外
    throw new InterruptedException();
  }
 } finally {
  if (failed)
   cancelAcquire(node);
 }
}

中斷例外拋出后，將會執行finally 代碼塊，取消正在進行嘗試獲取鎖的節點，

到此為止，lock()獲取鎖的概要程序為

先嘗試獲取鎖、失敗就將執行緒封裝成節點并加入到佇列尾部、進入自旋鎖（第一次嘗試獲取鎖失敗，將前驅節點的waitStatus改為-1；第二次嘗試獲取鎖失敗，因為前驅節點的waitStatus=-1，所以執行阻塞當前執行緒操作避免死回圈耗費資源）、等待頭節點執行緒釋放同步狀態之后，將發起解除阻塞指令或阻塞執行緒被中斷后，后繼節點再次嘗試獲取鎖，

取消例外節點

前面提到，在AQS#shouldParkAfterFailedAcquire(pred, node) 中談到，當節點waitStatus>0 時，也即是對帶有取消狀態的節點進行移除，那么節點在什么時候被改為CANCELLED 的呢？

在AQS#acquireQueued(node, arg)中，正常獲取到鎖時，failed都等于false，只有當發生例外時，failed 才等于true，從而執行到AQS#cancelAcquire(node)，也就是說cancelAcquire() 是用于處理獲取鎖程序中，對發生例外節點的進行移除，

final boolean acquireQueued(final Node node, int arg) {
    // 是否發生例外的標志
    boolean failed = true;
    try {
        ......
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                ......
                failed = false; // 只要當前節點node正常獲取到鎖，就不會執行finally的cancelAcquire(node)
                return interrupted;
            }
            ......
        }
    } finally {
        /* 正常情況：只有return時，才會執行finally代碼，而只要return，failed都等于false，
         * 所以，failed 是為了避免節點發生例外時，node沒有被移除，
         */
        if (failed)
            cancelAcquire(node);
    }
}

詳細看下AQS#cancelAcquire(node) 是怎么處理的

// java.util.concurrent.locks.AbstractQueuedSynchronizer
/**
 * 取消正在進行嘗試獲取鎖的節點
 */
private void cancelAcquire(Node node) {
    // 如果節點不存在，則忽略
    if (node == null)
        return;
    // 使當前節點變成虛節點
    node.thread = null;

    // 跳過帶有“取消狀態”的前驅節點
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    // predNext節點
    Node predNext = pred.next;

    // 修改當前節點的waitStatus為CANCELLED=1
    node.waitStatus = Node.CANCELLED;

    // 如果當前節點為尾節點tail，則只需要移除自己即可，
    if (node == tail && compareAndSetTail(node, pred)) {
        // pred節點變成了尾節點tail，所以 pred.next=null
        compareAndSetNext(pred, predNext, null);
    } else {
        int ws;
        /* 當前節點的前驅節點不為頭節點，則true；
         * 前驅節點的ws狀態為SIGNAL，則true；ws不為SIGNAL，但ws<=0時（即不是取消狀態），則CAS操作改為SIGNAL，改成功則為true；
         * 前驅節點不是虛接點，則true
         */ 
        if (pred != head
            && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL)))
            && pred.thread != null) {
            /* 如果上述都滿足，則將“當前節點的前驅節點的后繼節點”指向“當前節點的后繼節點”
             * 說白了，節點的next指向就是由 A->B->C 改為 A->C；B為當前節點，
             * 關于節點的prev指向就沒有變 A<-B<-C 還是 A<-B<-C，也就是說B節點還沒真正斷開，
             * 節點的prev指向的修改需要判斷ws狀態是否為CANCELLED后，才做修改，
             */
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
            // 喚醒當前節點的后繼節點的阻塞執行緒
            unparkSuccessor(node);
        }
        // 當前節點的后繼節點指向自己
        node.next = node;
    }
}

以上都是對next節點的指向做修改，關于節點的prev指向的修改需要判斷ws狀態是否為CANCELLED后，才做修改，回圈向前查找取消節點，把取消節點從佇列中剔除，其對應原始碼如下

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 獲取前驅節點的等待狀態
    int ws = pred.waitStatus;
        ......
    if (ws > 0) {
        // 前驅節點被標上取消標志了，需要跳過前驅節點并不斷重試
        do {
            // 回圈向前查找取消節點，把取消節點從佇列中剔除
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        ......
    }
    return false;
}

在AQS#unparkSuccessor() 通過執行緒阻塞工具類方法LockSupport.unpark(thread) 來喚醒后繼節點的阻塞執行緒，AQS#unparkSuccessor() 除了取消例外節點時用到外，還在鎖的釋放時呼叫，實作功能都是--喚醒當前節點的后繼節點的阻塞執行緒，后繼節點就會繼續執行自旋鎖來嘗試獲取鎖，

// java.util.concurrent.locks.AbstractQueuedSynchronizer

/**
 * 喚醒當前節點的后繼節點的阻塞執行緒(如果存在)
 */
private void unparkSuccessor(Node node) {
    /*
     * 如果waitStatus<0,則將waitStatus 置為默認值0
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * 但如果為null 或為取消狀態，則從tail向前遍歷以查找到實際未取消的后繼節點，
     */
    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);
}

節點執行緒釋放鎖

獲取鎖搞懂后，釋放鎖就是很簡單了

處理流程

鎖在釋放時呼叫的關鍵流程：ReentrantLock#unlock() -> Sync#release(1) -> AQS#tryRelease(1) -> LockSupport#unparkSuccessor(head)

unparkSuccessor(head)方法在前面已經講述過不再贅述，前三個方法原始碼如下??

// java.util.concurrent.locks.ReentrantLock
public void unlock() {
    // ReentrantLock API 交由同步佇列模板方法實作
    sync.release(1);
}

// java.util.concurrent.locks.AbstractQueuedSynchronizer
public final boolean release(int arg) {
    // 嘗試釋放鎖，成功則喚醒后繼節點
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            // 喚醒當前節點的后繼節點的阻塞執行緒
            unparkSuccessor(h);
        return true;
    }
    return false;
}


// java.util.concurrent.locks.ReentrantLock.Sync
protected final boolean tryRelease(int releases) {
    // state：每釋放1次鎖就會-1（相反：重入性，每獲取1次鎖就會+1）
    int c = getState() - releases;
    // 當前執行緒是否是獨占鎖執行緒，不是就拋出例外
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    // 當state=0時，說明獲取鎖的次數已經釋放完，可以解除獨占鎖執行緒
    if (c == 0) {
        // 鎖釋放成功
        free = true;
        // 獨占鎖執行緒置為null
        setExclusiveOwnerThread(null);
    }
    // 記錄每次state的變化
    setState(c);
    return free;
}

最后附上以 ReentrantLock 的lock()為例，里面幾憾訓出了獲取鎖的所有代碼的執行程序