[Java併發] AQS抽象隊列同步器源碼解析--鎖獲取過程

要深刻了解java併發知識，AbstractQueuedSynchronizer（AQS）是必需要拿出來深刻學習的，AQS能夠說是貫穿了整個JUC併發包，例如ReentrantLock，CountDownLatch，CyclicBarrier等併發類都涉及到了AQS。接下來就對AQS的實現原理進行分析。

在開始分析以前，勢必先將CLH同步隊列了解一下

CLH同步隊列

CLH自旋鎖: CLH(Craig, Landin, and Hagersten locks): 是一個自旋鎖，能確保無飢餓性，提供先來先服務的公平性。CLH自旋鎖是一種基於隱式鏈表（節點裏面沒有next指針）的可擴展、高性能、公平的自旋鎖，申請線程只在本地變量上自旋，它不斷輪詢前驅的狀態，若是發現前驅釋放了鎖就結束自旋。

AQS中的CLH同步隊列：AQS中CLH同步隊列是對CLH自旋鎖進行了優化，其主要從兩方面進行了改造：節點的結構與節點等待機制。

1.在結構上引入了頭結點和尾節點，他們分別指向隊列的頭和尾，嘗試獲取鎖、入隊列、釋放鎖等實現都與頭尾節點相關，而且每一個節點都引入前驅節點和後後續節點的引用；

2.在等待機制上由原來的自旋改爲阻塞喚醒。

源碼中CLH的簡單表示

*      +------+  prev +-----+       +-----+
* head |      | <---- |     | <---- |     |  tail
*      +------+       +-----+       +-----+

Node就是實現CLH同步隊列的數據結構,計算下就瞭解下該類的相關字段屬性

AQS中重要的內部類Node

static final class Node {
    // 共享模式
    static final Node SHARED = new Node();
    // 獨佔模式
    static final Node EXCLUSIVE = null;

    // 若是屬性waitStatus == Node.CANCELLED,則代表該節點已經被取消
    static final int CANCELLED =  1;
    // 若是屬性waitStatus == Node.SIGNAL,則代表後繼節點等待被喚醒
    static final int SIGNAL    = -1;
    // 若是屬性waitStatus == Node.CONDITION,則代表是Condition模式中的節點等待條件喚醒
    static final int CONDITION = -2; 
    // 若是屬性waitStatus == Node.PROPAGATE，在共享模式下，傳播式喚醒後繼節點
    static final int PROPAGATE = -3; 
    // 用於標記當前節點的狀態，取值爲1，-1，-2，-3，分別對應以上4個取值
    volatile int waitStatus;
    // 前驅節點
    volatile Node prev;
    // 後繼節點
    volatile Node next;
    // 當前節點對應的線程，阻塞與喚醒的線程
    volatile Thread thread;
    // 使用Condtion時（共享模式下）的後繼節點，在獨佔模式中不會使用
    Node nextWaiter;
    final boolean isShared() {
            return nextWaiter == SHARED;
    }
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    Node() {    // Used to establish initial head or SHARED marker
    }

    Node(Thread thread, Node mode) {     // Used by addWaiter
        this.nextWaiter = mode;
        this.thread = thread;
    }

    Node(Thread thread, int waitStatus) { // Used by Condition
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

下面就開始着重對AQS中的重要方法進行分析說明

獲取鎖

1.acquire 開始獲取鎖

public final void acquire(int arg) {
    //若是tryAcquire返回true，即獲取到鎖就中止執行，不然繼續向下執行向同步隊列尾部添加節點
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

tryAcquire是用於獲取鎖的方法，在AQS中默認沒有實現具體邏輯，由子類自定義實現。

若是返回true則說明獲取到鎖，不然須要將當前線程封裝爲Node節點添加到同步隊列尾部。

2.當前節點入隊列

將當前執行的線程封裝爲Node節點並加入到隊尾

private Node addWaiter(Node mode) {// 首先嚐試快速添加到隊尾，失敗再正常執行添加到隊尾
    Node node = new Node(Thread.currentThread(), mode);
    // 快速方式嘗試直接添加到隊尾
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 若是快速添加到隊尾失敗則執行enq(node)添加到隊尾
    enq(node);
    return node;
}

enq方法循環遍歷添加到隊尾

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;
            }
        }
    }
}

addWaiter(Node mode)方法執行完後，接下來執行acquireQueued方法， 返回的是該線程是否須要中斷，該方法也是不停地循環獲取鎖，若是前節點是頭節點，則嘗試獲取鎖，獲取鎖成功則返回是否須要中斷標誌，若是獲取鎖失敗，則判斷是否須要阻塞並阻塞線程

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        // 標記是否須要被中斷
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 若是前驅節點是頭節點，而且獲取鎖成功，則返回
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 判斷獲取鎖失敗後是否須要阻塞當前線程，若是阻塞線程後再判斷是否須要被中斷線程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

shouldParkAfterFailedAcquire(p, node)方法判斷是否須要阻塞當前線程

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    // 若是ws == Node.SIGNAL，則說明當前線程已經準備好被喚醒，所以如今能夠被阻塞，以後等待被喚醒
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
        // 若是ws > 0,說明當前節點已經被取消，所以循環剔除ws>0的前驅節點
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        //若是ws<=0,則將標誌位設置爲Node.SIGNAL，當還不可被阻塞，須要的等待下次執行shouldParkAfterFailedAcquire判斷是否須要阻塞
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

若是shouldParkAfterFailedAcquire(p, node)方法返回true，說明須要阻塞當前線程，則執行parkAndCheckInterrupt方法阻塞線程，並返回阻塞過程當中線程是否被中斷

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this); // 阻塞線程，等待unpark()或interrupt()喚醒本身
    // 線程被喚醒後查看是否被中斷過。
    return Thread.interrupted();
}

那麼從新回到獲取鎖的方法acquire方法，若是acquireQueued(final Node node, int arg)返回true，也便是阻塞過程當中線程被中斷，則執行中斷線程操做selfInterrupt()

public final void acquire(int arg) {
    //若是tryAcquire返回true，即獲取到鎖就中止執行，不然繼續向下執行向同步隊列尾部添加節點，而後判斷是否被中斷過，是則執行中斷
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

中斷當前線程

static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

小結

AQS獲取鎖的過程：

1.執行tryAcquire方法獲取鎖，若是獲取鎖成功則直接返回，不然執行步驟2

2.執行addWaiter方法將當前線程封裝位Node節點並添加到同步隊列尾部，執行步驟3

3.執行acquireQueued循環嘗試獲取鎖，，若是獲取鎖成功，則判斷返回中斷標誌位，若是獲取鎖失敗則調用shouldParkAfterFailedAcquire方法判斷是否須要阻塞當前線程，若是須要阻塞線程則調用parkAndCheckInterrupt阻塞線程，並在被喚醒後再判斷再阻塞過程當中是否被中斷過。

4.若是acquireQueued返回true,說明在阻塞過程當中線程被中斷過，則執行selfInterrupt中斷線程

好了，以上就是AQS的鎖獲取過程，關於鎖釋放的分析會在後續繼續輸出。