併發編程-深刻淺出AQS

時間 2019-11-20

標籤併發編程深刻 aqs 简体版

原文原文鏈接

AQS是併發編程中很是重要的概念，它是juc包下的許多併發工具類，如CountdownLatch，CyclicBarrier，Semaphore 和鎖, 如ReentrantLock， ReaderWriterLock的實現基礎，提供了一個基於int狀態碼和隊列來實現的併發框架。本文將對AQS框架的幾個重要組成進行簡要介紹，讀完本文你將get到如下幾個點：java

AQS進行併發控制的機制是什麼node
AQS獨佔和共享模式是如何實現的編程
同步隊列和條件等待隊列的區別，和數據出入隊原則併發

一，AQS基本概念

AQS（AbstractQueuedSynchronizer）是用來構建鎖或者其餘同步組件的基礎框架，它使用了一個int成員變量來表示狀態，經過內置的FIFO（first in，first out）隊列來完成資源獲取線程的排隊工做。框架

隊列可分爲兩種，一種是同步隊列，是程序執行入口出處的等待隊列；而另外一種則是條件等待隊列，隊列中的元素是在程序執行時在某個條件上發生等待。工具

1.1 獨佔or共享模式

AQS支持兩種獲取同步狀態的模式既獨佔式和共享式。顧名思義，獨佔式模式同一時刻只容許一個線程獲取同步狀態，而共享模式則容許多個線程同時獲取。ui

1.2 同步隊列

當一個線程嘗試獲取同步狀態失敗時，同步器會將這個線程以及等待狀態等信息構形成一個節點加入到等待隊列中，同時會阻塞當前線程，當同步狀態釋放時，會把首節點中的線程喚醒，使其再次嘗試重複獲取同步隊列。線程

1.3 條件隊列

AQS內部類ConditionObject來實現的條件隊列，當一個線程獲取到同步狀態，可是卻經過Condition調用了await相關的方法時，會將該線程封裝成Node節點並加入到條件隊列中，它的結構和同步隊列相同。code

二，獨佔or共享模式

AQS框架中，經過維護一個int類型的狀態，來進行併發控制，線程一般經過修改此狀態信息來代表當前線程持有此同步狀態。AQS則是經過保存修改狀態線程的引用來實現獨佔和共享模式的。blog

/**
 * 獲取同步狀態
 */
public final void acquire(int arg) {
    //嘗試獲取同步狀態， 若是嘗試獲取到同步狀態失敗，則加入到同步隊列中
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
/**
 * 嘗試獲取同步狀態【子類中實現】，由於aqs基於模板模式，僅提供基於狀態和同步隊列的實 
 * 現思路，具體的實現由子類決定
 */
protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 若是當前狀態值爲0，而且等待隊列中沒有元素，執行修改狀態值操做
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            // 修改狀態值成功，記錄當前持有同步狀態的線程信息
            setExclusiveOwnerThread(current);
            return true;
        }
        // 若是當前線程已經持有同步狀態，繼續修改同步狀態【重入鎖實現原理】
    } else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

/**
 * 根據傳入的模式以及當前線程信息建立一個隊列的節點並加入到同步隊列尾部
 */
private Node addWaiter(Node mode) {
    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.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}
/**
 * 同步隊列中節點，嘗試獲取同步狀態
 */
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);
    }
}

獨佔式是如何控制得？

當修改狀態信息成功後，若是執行的是獨佔式操做，AQS的具體實現類中會保存當前線程的信息來聲明同步狀態已被當前線程佔用，此時其餘線程再嘗試獲取同步狀態會返回false。

三，同步隊列

3.1 隊列中保存那些信息？

同步隊列節點中主要保存着線程的信息以及模式(共享or獨佔)。

3.2 什麼時候執行入隊操做？

/**
 * 獲取同步狀態
 */
public final void acquire(int arg) {
    //嘗試獲取同步狀態， 若是嘗試獲取到同步狀態失敗，則加入到同步隊列中
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

複用上文中的代碼，不難看出再獲取同步狀態失敗後，會執行入隊操做。

3.3 什麼時候執行出隊操做？

當線程獲取同步狀態失敗時，會被封裝成Node節點加入到等待隊列中，此時全部節點都回進入自旋過程，首先判斷本身prev是否時頭節點，若是是則嘗試獲取同步狀態。
被阻塞線程的喚醒主要以靠前驅節點的出隊或阻塞線程被中斷來實現。

/**
 * 同步隊列中節點，嘗試獲取同步狀態
 * 
 * 1. 當一個線程獲取到同步狀態時，會將當前線程構造程Node並設置爲頭節點
 * 2. 並將原始的head節點設置爲null，以便於垃圾回收
 */
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);
    }
}

四，條件等待隊列

條件變量（ConidtionObject）是AQS中的一個內部類，用來實現同步隊列機制。同步隊列複用了等待隊列中Node節點，因此同步隊列到等待隊列中不須要進行額外的轉換。

4.1 何時執行入隊操做？

當線程獲取到同步狀態，可是在臨界區中調用了await()方法，此時該線程會被加入到對應的條件隊列彙總。
ps: 臨界區，加鎖和釋放鎖之間的代碼區域

/**
 * ConditionObject中的await方法，調用後使得當前執行線程加入條件等待隊列
 */
public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    // -----省略代碼------
}
/**
 * 添加等待線程
 */
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // -----省略代碼------
    // 將當前線程構造程條件隊列節點，並加入到隊列中
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

4.2 何時執行出隊操做？

當對應的Conditioni調用signial/signalAll()方法時回選擇從條件隊列中出隊列，同步隊列是經過自旋的方式獲取同步狀態，而條件隊列中的節點則經過通知的方式出隊。條件隊列中的節點被喚醒後會加入到入口等待隊列中。

/**
 * 喚醒當前條件等到隊列中的全部等待線程
 */
public final void signalAll() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignalAll(first);
}
/**
 * 遍歷隊列，將元素從條件隊列 加入到 同步隊列
 */
private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}
final boolean transferForSignal(Node node) {
    // -----省略代碼------
    // 執行入隊操做，將node添加到同步隊列中
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}