java併發編程學習之AQS

原理

全稱AbstractQueuedSynchronizer，當線程去獲取資源的時候，會根據狀態值state來判斷是否有鎖，若是有鎖，則加入到鏈表，鏈表裏的線程，經過自旋，判斷資源是否已經釋放，若是釋放，則獲取資源。

AQS結構

1. volatile Node head：阻塞的頭節點
2. volatile Node tail：阻塞的尾節點，新的阻塞節點加到最後
3. volatile int state：鎖的狀態，0說明未佔用，大於等於1說明已佔用
4. Thread exclusiveOwnerThread：佔用鎖的當前線程

node：雙向鏈表的節點信息

1. Node EXCLUSIVE：獨佔模式
2. volatile Thread thread：當前線程
3. volatile Node prev：前置節點
4. volatile Node next：後置節點
5. volatile int waitStatus：狀態字段，-1：等待被喚醒，大於0，被取消

源碼分析

這裏以非公平鎖爲例

加鎖

lock

獲取鎖

final void lock() {
    if (compareAndSetState(0, 1))//若是狀態是0，則設置爲1
        setExclusiveOwnerThread(Thread.currentThread());//設置佔用鎖爲當前的線程
    else
        acquire(1);//資源被佔用
}

acquire

鎖被佔用後，再嘗試獲取，獲取不到，進入阻塞隊列

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

tryAcquire：

調用的是Sync的nonfairTryAcquire方法。

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();//獲取當前線程
    int c = getState();//獲取當前狀態
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {//爲當前線程，重入
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

addWaiter

把線程封裝成node，加入隊列

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;//若是cas操做成功，設置雙向鏈表
            return node;
        }
    }
    enq(node);
    return node;
}

enq

經過自旋的方式，加入到尾節點

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;//獲取尾節點
        if (t == null) { // 若是尾節點爲空，初始化頭節點和尾節點，地址爲同一個
            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 (;;) {
            final Node p = node.predecessor();//獲取前置節點
            if (p == head && tryAcquire(arg)) {//若是前置節點是head，而且獲取到了鎖
                setHead(node);//把當前節點設置到head上面
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //若是既不是隊頭，或者沒有搶過其餘線程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())//若是隊頭是喚醒的狀態，就用parkAndCheckInterrupt掛起
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

shouldParkAfterFailedAcquire

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;//喚醒，返回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 must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//前置節點狀態設置爲喚醒
    }
    return false;
}

喚醒

unlock

public void unlock() {
    sync.release(1);
}

release

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);//喚醒頭節點
        return true;
    }
    return false;
}

tryRelease

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;//可能重入的狀況
    if (Thread.currentThread() != getExclusiveOwnerThread())//非當前線程拋異常
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {//不用cas是由於僅有當前顯示有鎖
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);//若是頭節點當前waitStatus<0, 修改成0

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    Node s = node.next;
    //喚醒下一個節點，若是第一個爲空，從尾部遍歷上去，獲取最前面的waitStatus 小於0的節點
    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);//喚醒
}