java.util.concurrent包中的工具實現(xiàn)核心都是AQS，了解ReentrantLock的實現(xiàn)原理，需要先分析AQS以及AQS與ReentrantLock的關(guān)系。

這篇文章中分析了ReentrantLock#lock與ReentrantLock#unlock的實現(xiàn)，對于Condition的實現(xiàn)分析，另外文章再講，基本上大同小異。

ReentrantLock實現(xiàn)核心–AQS(AbstractQueuedSynchronizer)

AQS，隊列同步器，在juc包中的工具類都是依賴于AQS來實現(xiàn)同步控制，看一張AQS的結(jié)構(gòu)圖。

同步控制中主要使用到的信息如上圖所示。AQS可以被當(dāng)做是一個同步監(jiān)視器的實現(xiàn)，并且具有排隊功能。當(dāng)線程嘗試獲取AQS的鎖時，如果AQS已經(jīng)被別的線程獲取鎖，那么將會新建一個Node節(jié)點，并且加入到AQS的等待隊列中，這個隊列也由AQS本身自己維護。當(dāng)鎖被釋放時，喚醒下一個節(jié)點嘗試獲取鎖。

• 變量exclusiveOwnerThread在互斥模式下，表示當(dāng)前持有鎖的線程。
• 變量tail指向等待獲取AQS的鎖的節(jié)點隊列的最后一個
• 變量head指向隊列中head節(jié)點，head節(jié)點信息為空，不表示任何正在等待的線程。
• 變量state表示AQS同步器的狀態(tài)，在不同情況下含義可能不太一樣，例如以下幾種
  • 在ReentrantLock中，表示AQS的鎖是否已經(jīng)被占用獲取，0：沒有，>=1：已被獲取,當(dāng)大于1時表示被同一線程多次重入鎖。
  • 在CountDownLatch中，表示計數(shù)還剩的次數(shù)，當(dāng)?shù)竭_0時，喚醒等待線程。
  • 在Semaphore中，表示AQS還可以被獲取鎖的次數(shù)，獲取一次就減1，當(dāng)?shù)竭_0時，嘗試獲取的線程將會阻塞。

Node結(jié)構(gòu)

Node節(jié)點是AQS管理的等待隊列的節(jié)點元素，除了head節(jié)點之外，其他一個節(jié)點就代表一個正在等待線程的隊列。Node一般的重要參數(shù)有幾個。

• prev 前置節(jié)點
• next后置節(jié)點
• thread 代表的線程
• waitStatus節(jié)點的等待狀態(tài)
  • 1表示節(jié)點已經(jīng)取消，也就是線程可能已經(jīng)中斷，不需要再等待獲取鎖了，在后續(xù)代碼中會處理跳過waitStatus等于1的節(jié)點
  • -1表示當(dāng)前節(jié)點的后置節(jié)點代表的線程需要被掛起
  • -2表示當(dāng)前線程正在等待的是Condition鎖

ReentrantLock實現(xiàn)分析

二者關(guān)聯(lián)

ReentrantLock實現(xiàn)核心是基于AQS,看下面一張圖，分析AQS與ReentrantLock的關(guān)系。

從圖中可以看出，ReentrantLock里面有最終兩個內(nèi)部類，F(xiàn)airSync和NonfairSync通過繼承AbstractQueuedSynchronizer的功能，來實現(xiàn)兩種同步互斥方案：公平鎖和非公平鎖。

在ReentrantLock中最終lock和unlock操作，都由FairSync和NonfairSync實際完成。

NonfairSync分析

下面看一個最簡單利用ReentrantLock實現(xiàn)互斥的例子。

        ReentrantLock lock = new ReentrantLock();
		//嘗試獲取鎖
        lock.lock();
        //獲得鎖后執(zhí)行邏輯......
        //......
		//解鎖
        lock.unlock();

在ReentrantLock的默認無參構(gòu)造方法中，創(chuàng)建的是非公平鎖

    public ReentrantLock() {
        sync = new NonfairSync();
    }

下面分析lock.lock();這句代碼是如何實現(xiàn)同步互斥的。

NonfairSync#lock

點開lock.lock();源碼，可以看到最終實際調(diào)用的是NonfairSync#lock，這是分析的入口。

NonfairSync#lock源碼如下。

final void lock() {
    if (compareAndSetState(0, 1))//【step1】
        setExclusiveOwnerThread(Thread.currentThread());//【step2】
    else
        acquire(1);//【step3】
}

【step1】上面有提到，NonfairSync繼承自AbstractQueuedSynchronizer，NonfairSync就是一個AQS，因此在步驟【1】其實就是利用CAS（一個原子性的比較并設(shè)置操作）嘗試設(shè)置AQS的state為1。如果設(shè)置成功，表示獲取鎖成功；如果設(shè)置失敗，表示state之前已經(jīng)是>=1，已經(jīng)被別的線程占用了AQS的鎖，所示無法設(shè)置state為1，稍后會把線程加入到等待隊列。

**非公平鎖與公平鎖：**對于NonfairSync非公平鎖來說，線程只要執(zhí)行l(wèi)ock請求，就會立馬嘗試獲取鎖，不會管AQS當(dāng)前管理的等待隊列中有沒有正在等待的線程，這種操作是不公平的，沒有先來后到；而稍后介紹的FairSync公平鎖，則會在lock請求進行時，先判斷AQS管理的等待隊列中是否已經(jīng)有正在等待的線程，有的話就是不嘗試獲取鎖，直接進入等待隊列，保證了公平性。

這一步需要熟悉的是CAS操作，分析一下compareAndSetState源碼，如下。這一步利用unsafe包的cas操作，unsafe包類似一種java留給開發(fā)者的后門，可以用來直接操作內(nèi)存數(shù)據(jù)，并且保證這個操作的原子性。在下面的代碼中，stateOffset表示state比變量的內(nèi)存偏移地址，用來尋找state變量內(nèi)存位置。整個cas操作就是找到內(nèi)存中當(dāng)前的state變量值，并且與expect期待值比較，如果跟期待值相同，那么表示這個值是可以修改的，此時就會對state值進行更新；如果與期待值不一樣，那么將不能進行更新操作。unsafe保證了比較與設(shè)置值的過程是原子性的，在這一步不會出現(xiàn)線程安全問題。

protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

【step2】操作是在設(shè)置state成功之后，表示當(dāng)前線程獲取AQS鎖成功，需要設(shè)置AQS中的變量exclusiveOwnerThread為當(dāng)前持有鎖的線程，做保存記錄。

【step3】當(dāng)嘗試獲取鎖失敗的時候，就需要進行步驟3，執(zhí)行acquire,進行再次嘗試或者線程進入等待隊列。

AbstractQueuedSynchronizer#acquire

當(dāng)?shù)谝淮螄L試獲取鎖失敗之后，執(zhí)行【step3】acquire方法，這個方法可以講一個嘗試獲取鎖失敗的線程放入AQS管理的等待隊列進行等待，并且將線程掛起。實現(xiàn)邏輯在AbstractQueuedSynchronizer實現(xiàn)，NonfairSync通過繼承AbstractQueuedSynchronizer獲得等待隊列管理的功能。

下面看源碼。

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

NonfairSync#tryAcquire–鎖重入實現(xiàn)

首先，執(zhí)行tryAcquire再次嘗試一次獲取lock，tryAcquire是由子類實現(xiàn)，也就是這里調(diào)用NonfairSync#tryAcquire方法，跟蹤調(diào)用，最終執(zhí)行代碼如下。

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        //如果此時state已經(jīng)變?yōu)?，再次嘗試一次獲取鎖
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        //否則判斷當(dāng)前持有AQS的鎖的線程是不是當(dāng)前請求獲取鎖的線程，是的話就進行鎖重入。
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

對于NonfairSync#tryAcquire的實現(xiàn)，首先是重新嘗試一次獲取鎖，如果還是獲取不到，就嘗試判斷當(dāng)前的是不是屬于重入鎖的情形。

對于重入鎖的情形，就需要對state進行累加1，意思就是重入一次就對state加1。這樣子，在解鎖的時候，每次unlock就對state減一，等到state的值為0的時候，才能喚醒下一個等待線程。

如果獲取成功，返回true；否則返回false，繼續(xù)執(zhí)行下一步acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，添加一個Node節(jié)點進入AQS管理的等待隊列。

AbstractQueuedSynchronizer#addWaiter–添加Node到等待隊列

這個方法屬于隊列管理，也是由AbstractQueuedSynchronizer默認實現(xiàn)，NonfairSync繼承獲得。

查看addWaiter源碼實現(xiàn)。

private Node addWaiter(Node mode) {
    //新建一個Node節(jié)點，mode傳入表示當(dāng)前線程之間競爭是互斥模式
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    //嘗試添加當(dāng)前新建Node到鏈表隊列末尾
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        //利用cas設(shè)置尾指針指向的節(jié)點為當(dāng)前線新建節(jié)點
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //當(dāng)前是空的沒有任何正在等待的線程Node的時候，執(zhí)行enq(node)，初始化等待隊列
    enq(node);
    return node;
}

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            //新建一個空的頭節(jié)點
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            //跟之前一樣，利用cas這只當(dāng)前新建節(jié)點為尾節(jié)點
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

以上操作，完成將一個節(jié)點加入隊列操作。加入完成之后，返回這個新加入的節(jié)點Node給acquireQueued方法繼續(xù)執(zhí)行。

acquireQueued(addWaiter(Node.EXCLUSIVE), arg))–鎖競爭優(yōu)化

上面既然都完成了等待節(jié)點如隊列的操作，為什么不直接掛起線程進入等待呢？因此這里的設(shè)計者做了一個優(yōu)化操作。acquireQueued方法其實就是為了減少線程掛起、喚醒次數(shù)而作的優(yōu)化操作。

下面看看acquireQueued的代碼實現(xiàn)。

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            //獲取當(dāng)前競爭鎖的線程Node的前置節(jié)點
            final Node p = node.predecessor();
            //【step1】
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //【step2】
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

【step1】假如前置節(jié)點是head，那么表示當(dāng)前線程是等待隊列中最大優(yōu)先級的等待線程，可以繼續(xù)最后的嘗試獲取鎖，因為很有可能會獲取到鎖，從而避免線程掛起、喚醒，耗費資源，這里也算是最終一次嘗試獲取。

【step2】shouldParkAfterFailedAcquire(p, node)是檢查當(dāng)前是否有必要掛起，前面我們說過，只有當(dāng)前置節(jié)點的waitStatus是-1的時候才會掛起當(dāng)前節(jié)點代表的線程。當(dāng)shouldParkAfterFailedAcquire(p, node)返回true的時候，就可以執(zhí)行parkAndCheckInterrupt()來掛起線程。

shouldParkAfterFailedAcquire(p, node)源碼如下。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        //前置節(jié)點是-1，返回true表示線程可掛起
        return true;
    if (ws > 0) {
        //前置節(jié)點大于0表示前置節(jié)點已經(jīng)取消，那么進行跳過前置節(jié)點的操作，做鏈表的基本刪除節(jié)點操作
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        //如果前置節(jié)點還是0,表示前置節(jié)點Node的waitStatus是初始值，需要設(shè)置為-1，然后外層循環(huán)重新執(zhí)行shouldParkAfterFailedAcquire方法，即可掛起當(dāng)前線程。
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

private final boolean parkAndCheckInterrupt() {
    //阻塞掛起線程，等待喚醒
    LockSupport.park(this);
    //喚醒后，重置中斷標(biāo)記，線程中斷標(biāo)記位為不中斷
    return Thread.interrupted();
}

FairSync分析

之前提到

**非公平鎖與公平鎖：**對于NonfairSync非公平鎖來說，線程只要執(zhí)行l(wèi)ock請求，就會立馬嘗試獲取鎖，不會管AQS當(dāng)前管理的等待隊列中有沒有正在等待的線程，這種操作是不公平的，沒有先來后到；而稍后介紹的FairSync公平鎖，則會在lock請求進行時，先判斷AQS管理的等待隊列中是否已經(jīng)有正在等待的線程，有的話就是不嘗試獲取鎖，直接進入等待隊列，保證了公平性。

所以兩者的實現(xiàn)區(qū)別在于第一次嘗試lock的動作不一樣。

FairSync#lock

final void lock() {
    acquire(1);
}

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

最終差異體現(xiàn)在FairSync#tryAcquire的實現(xiàn)（第一次嘗試獲取lock）

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        //hasQueuedPredecessors判斷隊列是否還有別的線程在等待鎖，沒有的話就嘗試獲取lock
        //如果有別的線程在等待鎖，就不會嘗試獲取lock；下面如果也不是重入的情況的話就直接進入等待隊列
        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;
}

AbstractQueuedSynchronizer#release --AQS解鎖操作

AQS中定義了解鎖操作的模板方法，tryRelease(arg)是不同AQS子類實現(xiàn),對state的多樣化操作。例如ReentrantLock中的tryRelease(arg)操作比較明顯的就是對state減一。

tryRelease(arg)返回結(jié)果表示本次操作后是否需要喚醒下一個等待節(jié)點。對于ReentrantLock就是state減一之后是否變?yōu)?了。如果需要喚醒下一個節(jié)點的線程，那么判斷一下Head有沒有下一個要喚醒的節(jié)點線程，有的話就進行喚醒操作unparkSuccessor(h);

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

ReentrantLock.Sync#tryRelease–解鎖實現(xiàn)

看一下ReentrantLock.Sync的tryRelease實現(xiàn).是如何為state減一 的。。

protected final boolean tryRelease(int releases) {
    //獲取state減掉realease，對于ReentrantLock就是默認減一
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        //如果減到0了，那么久釋放鎖
        free = true;
        //設(shè)置持有線程為null
        setExclusiveOwnerThread(null);
    }
    //設(shè)置state為新的
    setState(c);
    return free;
}

至于這里設(shè)置state為啥不同cas操作，因為

    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();

所以永遠只有持有鎖的線程會做解鎖減一的操作，state設(shè)置是線程安全的。

注意一下

其實這里還沒分析Condition的實現(xiàn)原理，篇幅太長，下次另開文章分析。以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論