Java隊列同步器（AQS）究竟是怎麼一回事

時間  2019-11-07
標籤 java 隊列 同步 aqs 究竟 怎麼 一回事 欄目 Java 简体版
原文   原文鏈接

0 簡介

隊列同步器AbstractQueuedSynchronizer（後面簡稱AQS）是實現鎖和有關同步器的一個基礎框架。java

在JDK5中，Doug Lea在併發包中加入了大量的同步工具，例如重入鎖（ReentrantLock）、讀寫鎖（ReentrantReadWriteLock）、信號量（Semaphore）、CountDownLatch等，都是基於AQS的。node

其內部經過一個被標識爲volatile的名爲state的變量來控制多個線程之間的同步狀態。多個線程之間能夠經過AQS來獨佔式或共享式的搶佔資源。編程

基於AQS，能夠很方便的實現Java中不具有的功能。安全

例如，在鎖這個問題上，Java中提供的是synchronized關鍵字，用這個關鍵字能夠很方便的實現多個線程之間的同步。但這個關鍵字也有不少缺陷，好比：微信

  • 他不支持超時的獲取鎖，一個線程一旦沒有從synchronized上獲取鎖，就會卡在這裏，沒有機會逃脫。因此一般由synchronized形成的死鎖是無解的。
  • 不可響應中斷。
  • 不能嘗試獲取鎖。若是嘗試獲取時沒獲取到，馬上返回，synchronized不具有這一特性。

而ReentrantLock基於AQS將上述幾點都作到了。併發

1 核心結構

從AbstractQueuedSynchronizer的名字能夠看出，AQS中必定是基於隊列實現的（Queue）。在AQS內部，是經過鏈表實現的隊列。鏈表的每一個元素是其內部類Node的一個實現。而後AQS經過實例變量head指向隊列的頭，經過實例變量tail指向隊列的尾。框架

其源碼定義以下：工具

/** * Head of the wait queue, lazily initialized. Except for * initialization, it is modified only via method setHead. Note: * If head exists, its waitStatus is guaranteed not to be * CANCELLED. */
private transient volatile Node head;

/** * Tail of the wait queue, lazily initialized. Modified only via * method enq to add new wait node. */
private transient volatile Node tail;

/** * The synchronization state. */
private volatile int state;


static final class Node {

	/** 標識爲共享式 */
    static final Node SHARED = new Node();
    /** 標識爲獨佔式 */
    static final Node EXCLUSIVE = null;

	/** 同步隊列中等待的線程等待超時或被中斷，須要從等待隊列中取消等待，進入該狀態的節點狀態將再也不變化 */
    static final int CANCELLED =  1;

    /** 當前節點的後繼節點處於等待狀態，且當前節點釋放了同步狀態，須要經過unpark喚醒後繼節點，讓其繼續運行 */
    static final int SIGNAL    = -1;

    /** 當前節點等待在某一Condition上，當其餘線程調用這個Conditino的signal方法後，該節點將從等待隊列恢復到同步隊列中，使其有機會獲取同步狀態 */
    static final int CONDITION = -2;

    /** 表示下一次共享式同步狀態獲取狀態將無條件的傳播下去 */
    static final int PROPAGATE = -3;

	/* 當前節點的等待狀態，取值爲上述幾個常量之一，另外，值爲0表示初始狀態 */
    volatile int waitStatus;

    /* 前驅節點 */
    volatile Node prev;

    /* 後繼節點 */
    volatile Node next;

    /* 等待獲取同步狀態的線程 */
    volatile Thread thread;

    /* 等待隊列中的後繼節點 */
    Node nextWaiter;
    
    // ...
}
複製代碼

當線程經過AQS獲取同步狀態時，AQS會將當前線程封裝到Node內部，併入隊。因此在多個線程併發獲取同步狀態時，AQS內部會持有以下結構的隊列：ui

AQS內部隊列

下文會基於這個隊列模型，說明一下線程在AQS中獲取同步狀態時的原理。this

2 實現原理

從AQS的名字能夠看出來，做者是但願AQS做爲一個基類來向外提供服務的（以Abstract標識）。因此一般AQS是以繼承的方式使用的。

AQS提供了幾個模板方法供實現類本身實現定製功能。

這幾個方法是：

  • boolean tryAcquire(int arg)：獨佔式的獲取同步狀態，一般經過以CAS的方式修改state的值來實現特定功能。
  • boolean tryRelease(int arg)：獨佔式的釋放同步狀態，一般也是修改state的值。
  • int tryAcquireShared(int arg)：共享式的獲取同步狀態，返回值>=0表示成功，不然失敗。
  • boolean tryReleaseShared(int arg)：共享式的釋放同步狀態，一樣經過修改state值來實現。
  • boolean isHeldExclusively()：表示AQS是否被當前線程獨佔。

這幾個方法的默認實現都會拋出UnsupportedOperationException異常。

目前咱們不用關心這幾個方法，只要明白其內部是經過控制state的值來管理同步狀態便可。

2.1 同步狀態的獲取

一般，實現類會優先嚐試修改state的值，來獲取同步狀態。例如，若是某個線程成功的將state的值從0修改成1，表示成功的獲取了同步狀態。 這個修改的過程是經過CAS完成的，因此能夠保證線程安全。

反之，若是修改state失敗，則會將當前線程加入到AQS的隊列中，並阻塞線程。

AQS內部提供了三個方法來修改state的狀態，其源碼以下：

/** * Returns the current value of synchronization state. * This operation has memory semantics of a {@code volatile} read. * @return current state value */
protected final int getState() {
    return state;
}

/** * Sets the value of synchronization state. * This operation has memory semantics of a {@code volatile} write. * @param newState the new state value */
protected final void setState(int newState) {
    state = newState;
}

/** * Atomically sets synchronization state to the given updated * value if the current state value equals the expected value. * This operation has memory semantics of a {@code volatile} read * and write. * * @param expect the expected value * @param update the new value * @return {@code true} if successful. False return indicates that the actual * value was not equal to the expected value. */
protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
複製代碼

2.2同步隊列

如上文所述，AQS內部其實是一個FIFO的雙端隊列，當線程獲取同步狀態失敗時，就會構建一個Node並添加到隊列尾部（此過程是線程安全的，CAS實現），並阻塞當前線程（經過LockSupport.park()方法）； 當釋放同步狀態時，AQS會先判斷head節點是否爲null，若是不是null，說明有等待同步狀態的線程，就會嘗試喚醒head節點，使其從新競爭同步狀態。

2.3 獨佔式同步狀態的獲取

獨佔式的意思就是說同一時間只能有一個線程得到同步狀態。

AQS會先嚐試調用實現類的tryAcquire方法獲取同步狀態，若是獲取失敗，會嘗試將其封裝爲Node節點添加到同步隊列尾部。

獨佔式同步狀態的獲取經過AQS的acquire方法實現。其源碼以下：

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

這個方法會先嚐試獲取一次同步狀態（tryAcquire），若是獲取失敗，會經過addWaiter方法將當前線程加入到同步隊列。 並在acquireQueued方法中將當前線程阻塞（LockSupport.park()），並進入自旋狀態，以獲取同步狀態。

下面具體看一下他是如何構建Node並將其添加到隊尾的。 首先是addWaiter：

/** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */
private Node addWaiter(Node mode) {
    // mode = Node.EXCLUSIVE，表示是獨佔模式
    Node node = new Node(Thread.currentThread(), mode);
    // 先快速的經過CAS的方式將Node添加到隊尾，若是失敗，再進入enq方法經過無限循環添加
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

/** * Inserts node into queue, initializing if necessary. See picture above. * @param node the node to insert * @return node's predecessor */
private Node enq(final Node node) {
    // 無限循環的將node添加到隊尾，保證能添加成功
    /* 注意：若是是首次向隊列中添加Node，那麼調addWaiter方法時，tail仍是null，因此addWaiter方法不會設置成功，會直接進入這個方法 進入這個方法後，因爲tail仍然是null，因此會走到第一個if裏面，這是會建立一個空的Node出來做爲頭結點 而後再次循環，此時tail不是null了，會進入else的代碼中，這時，纔會將須要add的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;
            }
        }
    }
}
複製代碼

再看下acquireQueued方法：

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 進入自旋，不斷的獲取同步狀態
        for (;;) {
            // 獲取node在隊列中的前驅節點
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                // 若是成功進入到這塊代碼，說明成功的獲取了同步狀態
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 獲取不成功，調用LockSupport.park()方法將當前線程阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
複製代碼

shouldParkAfterFailedAcquire方法用戶判斷是否須要阻塞當前線程，方法內會操做當前隊尾節點的前驅節點的waitStatus，並依據waitStatus判斷是否須要park。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL) // Node.SIGNAL == -1
        /* * 代表當前節點須要其餘線程的喚醒才能繼續執行，此時能夠安全的park。 */
        return true;
    if (ws > 0) {
        /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /* * 若是一個節點是初始狀態，即waitStatus=0時， * 將前驅節點的waitStatus設置爲-1，代表其須要別的線程喚醒才能繼續執行 */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
複製代碼

當shouldParkAfterFailedAcquire方法判斷當前節點須要被park時，會調用parkAndCheckInterrupt將其阻塞：

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}
複製代碼

2.4 獨佔式同步狀態的釋放

獨佔式的同步狀態釋放，在AQS中是經過release()方法實現的。此方法源碼以下：

public final boolean release(int arg) {
    // 嘗試調用實現類的tryRelease方法來修改同步狀態（state）
    if (tryRelease(arg)) {
        Node h = head;
        /* 1.若是head節點是null，表示沒有其餘線程競爭同步狀態，直接返回釋放成功 2.若是head節點不是null，代表有競爭。經過unparkSuccessor方法，經過unpark方法喚醒head節點的next節點。使其從新嘗試競爭同步狀態。 */
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
複製代碼

unparkSuccessor方法會喚醒head節點的next節點，使其能夠從新競爭同步狀態：

private void unparkSuccessor(Node node) {
    /* * 若是waitStatus的值是負數，例如：-1（等待signal） * 則將其值還原爲0 */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /* * 獲取頭結點的next節點，若是非空，則unpark他 */
    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);
}
複製代碼

2.5 獨佔式同步狀態獲取與釋放-圖示

下面會經過畫圖方式展現一下源碼中的過程，首先咱們假設tryAcquire的實現以下：

boolean tryAcquire(int acquires) {
    return compareAndSetState(0, acquires);
}
複製代碼

參數acquires固定傳1，意爲：經過CAS，若是成功將state的值從0修改成1，表示獲取同步狀態成功，不然失敗，須要加入同步隊列。

假設tryRelease的實現以下：

boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (c == 0) {
        setState(c);
        return true;
    }
    return false;
}
複製代碼

參數releases固定傳1，意爲：若是當前state-1=0，視爲釋放成功，其餘線程可競爭同步狀態。

假設有三個線程併發獲取同步狀態，標識爲t一、t二、t3，三個線程同時經過acquire方法修改state值。

假設t1修改爲功，t2和t3修改失敗。

t1修改爲功以後，將state值變爲1，並直接返回。此時head和tail都是空，因此同步隊列也是空的。此時同步隊列狀態以下：

t1成功修改state狀態

t2線程競爭同步狀態失敗，加入到同步隊列中：

t2加入同步隊列

t3線程競爭同步狀態失敗，加入到同步隊列中：

t3加入同步隊列

t1線程執行完畢，釋放資源。 先將state還原爲0，再unpark頭結點的next節點（t2節點），使之重獲同步狀態的競爭資格。

t1釋放同步狀態

假設t2被喚醒後成功的獲取到了同步狀態（即調用tryAcquire方法併成功將state設置爲1），t2會將本身所在的Node設置爲head節點，並將原head節點的next設置爲null（有助於GC）

t2從新獲取同步狀態

t2執行完成，釋放同步狀態，將state設置爲0，同時喚醒t3，使之再次具有競爭資格

t2釋放同步狀態

假設t3成功獲取同步狀態，此時t3將本身所在的Node設置爲head節點，並將以前的head節點的next設置爲null（即將t2的next設置爲null）

t3從新獲取同步狀態

t3執行完畢，釋放同步狀態，將state設置爲0。因爲此時其waitStatus等於0，表示已經沒有後繼節點須要unpark，直接返回釋放成功

t3釋放同步狀態

最後的t3節點並無被清空，由於他能夠用做下一次同步狀態競爭的head節點。

2.6 超時獲取同步狀態

tryAcquireNanos方法實現了這個功能。他與上面描述的獲取同步狀態的過程大體相同，只不過是加上了時間的判斷。 也就是說，每次自旋獲取同步狀態時，先判斷當前時間是否超過了指定的超時時間，若是超時直接返回獲取失敗。

下面來看下源碼，tryAcquireNanos方法源碼以下：

public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();

    // 先嚐試獲取同步狀態，若是失敗，嘗試超時獲取
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}
複製代碼

能夠發現，最終是doAcquireNanos方法實現的超時功能，這個方法中，大部分邏輯與上面的過程是一直的。 註釋中說明了有區別的地方。

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;
            // 沒超時的話，判斷剩餘時間是否大於1000納秒，若是大於才park當前線程
            // 不然，不park，直接進入下一次自旋獲取，由於這個時間足夠小了，可能已經超出了一次系統調用的時間
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold) // spinForTimeoutThreshold = 1000
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
複製代碼

3 參考

  • Java併發編程的藝術 方騰飛，魏鵬，程曉明
歡迎關注個人微信公衆號

公衆號
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程