一、類圖

ReentrantrantLock 的類圖如下所示，可以看到，它也是實(shí)現(xiàn)了 Lock 接口，和我們自定義不可重入鎖的步驟是一樣的，其內(nèi)部也是維護(hù)了一個(gè)內(nèi)部類同步器類 Sync，這個(gè)同步器類也是繼承自 AQS，不過需要注意的是這里面的同步器類 Sync 是抽象的，有兩個(gè)實(shí)現(xiàn)，一個(gè)是非公平鎖實(shí)現(xiàn) NonfairSync，一個(gè)是公平鎖實(shí)現(xiàn) FairSync。

二、非公平鎖實(shí)現(xiàn)原理

2.1 加鎖流程

先從構(gòu)造器開始看，默認(rèn)為非公平鎖實(shí)現(xiàn)，如下代碼，其中 NonfairSync 繼承自 AQS。

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

先看加鎖流程，如下的 lock() 方法，可以看到，方法內(nèi)部調(diào)用的是NonfairSync 實(shí)現(xiàn)的 lock() 方法

    public void lock() {
        sync.lock();
    }

接下來我們看下NonfairSync 的 lock() 方法是如何實(shí)現(xiàn)的，如下圖：

當(dāng)沒有競(jìng)爭(zhēng)的時(shí)候，通過 CAS 將 state 由 0 改成 1，并把 owner 線程改成自己，如下圖：

當(dāng)?shù)谝粋€(gè)競(jìng)爭(zhēng)出現(xiàn)時(shí)，如下圖，此時(shí)owner 已經(jīng)是 Thread0 了，此時(shí)來了一個(gè) Thread-1，此時(shí) Thread-1 還是執(zhí)行 lock() 方法。

此時(shí) Thread-1 的CAS 肯定是失敗的，此時(shí)就進(jìn)入 acquire() 方法，此方法的內(nèi)容如下所示：

1、進(jìn)入 tryAcquire 邏輯，這時(shí) state 已經(jīng)是 1，結(jié)果仍然失敗。

2、接下來進(jìn)入 addWaiter 邏輯，構(gòu)造 Node 隊(duì)列，如下圖，會(huì)在 head 后面添加一個(gè)節(jié)點(diǎn)，這個(gè)節(jié)點(diǎn)是雙向鏈表，即 head 指向一個(gè) node 節(jié)點(diǎn)，node 節(jié)點(diǎn)又指向下一個(gè) node 節(jié)點(diǎn)。

圖中黃色三角表示該 Node 的 waitStatus 狀態(tài)，其中 0 為默認(rèn)正常狀態(tài)，Node 的創(chuàng)建是懶惰的。

需要注意的是，首次創(chuàng)建 node 時(shí)會(huì)創(chuàng)建兩個(gè)，其中第一個(gè) Node 稱為 Dummy（啞元）或哨兵，用來占位，并不關(guān)聯(lián)線程。只有第二個(gè) Node 才會(huì)關(guān)聯(lián)線程 Thread-1，因?yàn)樗?jìng)爭(zhēng)失敗了，因?yàn)樗堑谝粋€(gè)加入的，所以它指向了鏈表的尾部。其實(shí)這個(gè)就是我們的等待隊(duì)列。

等到創(chuàng)建成功之后，當(dāng)前線程就會(huì)進(jìn)入到acquireQueued 邏輯，如下圖，這個(gè)方法主要是在一個(gè)死循環(huán)中不斷嘗試獲得鎖，失敗后進(jìn)入 park 阻塞。

1、首先獲取前驅(qū)節(jié)點(diǎn)，然后檢查前驅(qū)節(jié)點(diǎn)，判斷前驅(qū)階段是否為頭節(jié)點(diǎn)，如果是，說明 Thread-1 的這個(gè) Node 是處于第二位的，那么它就有資格再次調(diào)用 tryAcquire() 方法再次獲取鎖。

2、如果還是沒有獲取到鎖，此時(shí)就會(huì)進(jìn)入到shouldParkAfterFailedAcquire 邏輯，這個(gè)方法會(huì)把當(dāng)前線程節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)的 waitStatus 改為 -1，改為 -1 的意思是它有責(zé)任喚醒它的后繼節(jié)點(diǎn)，此時(shí)方法返回 false，如下圖：

3、shouldParkAfterFailedAcquire 執(zhí)行完畢回到 acquireQueued ，再次 tryAcquire 嘗試獲取鎖，當(dāng)然這時(shí)state 仍為 1，失敗。

4、當(dāng)再次進(jìn)入 shouldParkAfterFailedAcquire 時(shí)，這時(shí)因?yàn)槠淝膀?qū) node 的 waitStatus 已經(jīng)是 -1，這次返回 true。

5、此時(shí)就會(huì)進(jìn)入 parkAndCheckInterrupt()，即 Thread-1 被阻塞住，用灰顏色表示，如下圖：

再次有多個(gè)線程經(jīng)歷上述過程競(jìng)爭(zhēng)失敗，就會(huì)變成如下的樣子：

2.2 解鎖流程

解鎖流程，如下的 lock() 方法，可以看到，方法內(nèi)部調(diào)用的同步器 AQS 的 release() 方法。

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

release()方法的內(nèi)容如下所示：

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

假設(shè)此時(shí) Thread-0 釋放鎖，就會(huì)進(jìn)入到 tryRelease() 方法，如果返回 true，就證明釋放成功了，就會(huì)把owner 線程設(shè)置為 null，并且把 state 設(shè)置為 0，如下圖：

tryRelease() 方法返回 true 之后，還需要檢查隊(duì)列中的 head 是否為空，還需要檢查 head 中的 waitState 是否不等于 0 ，若返回 true ，則需要調(diào)用 unparkSuccessor() 方法喚醒下一個(gè)節(jié)點(diǎn)。

unparkSuccessor() 方法是找到隊(duì)列中離 head 最近的一個(gè) Node（沒取消的），unpark 恢復(fù)其運(yùn)行，本例中即為 Thread-1，此時(shí)回到 Thread-1 的acquireQueued() 方法的流程，如下圖：

如果加鎖成功（沒有競(jìng)爭(zhēng)），會(huì)設(shè)置 exclusiveOwnerThread 為 Thread-1，state = 1；head 指向剛剛 Thread-1 所在的 Node，該 Node 清空 Thread；原本的 head 因?yàn)閺逆湵頂嚅_，而可被垃圾回收。

如果這時(shí)候有其它線程來競(jìng)爭(zhēng)（非公平的體現(xiàn)），例如這時(shí)有 Thread-4 來了，Thread-4 并不是阻塞隊(duì)列里面的線程，如下圖：

如果不巧又被 Thread-4 占了先，那么 Thread-4 被設(shè)置為 exclusiveOwnerThread，state = 1；Thread-1 再次進(jìn)入 acquireQueued 流程，獲取鎖失敗，重新進(jìn)入 park 阻塞。

三、可重入原理

接下來我們看下非公平鎖的可重入的實(shí)現(xiàn)原理，我們看下他的獲取鎖和釋放鎖的相關(guān)代碼，

獲取鎖最終調(diào)用的是 nonfairTryAcquire() 方法，傳入的 acquires 就是 1，首先判斷 state 的狀態(tài)是否是 0，如果是 0 就代表還沒有別人獲取鎖，然后我就去 CAS ，如果成功則設(shè)置 owner 設(shè)置為自己，然后返回 true。

如果是同一個(gè)線程下次又調(diào)用nonfairTryAcquire() 方法，此時(shí)就回去判斷當(dāng)前線程是否為 owner 持有的線程，如果是，則證明發(fā)生了鎖重入，此時(shí)就是將 state 的值 ++，由 1 變成了 2，

static final class NonfairSync extends Sync {
    // ...

    // Sync 繼承過來的方法, 方便閱讀, 放在此處
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        // 如果已經(jīng)獲得了鎖, 線程還是當(dāng)前線程, 表示發(fā)生了鎖重入
        else if (current == getExclusiveOwnerThread()) {
            // state++
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

鎖釋放的時(shí)候調(diào)用的是 tryRelease() 方法，如下所示，此時(shí)就會(huì)把 state 的值 --，此時(shí)返回一個(gè) false，表示我只是讓 state -- ，并沒有真正的釋放鎖，直到下面的這個(gè) c 變成 0 了，才會(huì)釋放鎖，返回 true。

    // Sync 繼承過來的方法, 方便閱讀, 放在此處
    protected final boolean tryRelease(int releases) {
        // state--
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        // 支持鎖重入, 只有 state 減為 0, 才釋放成功
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

四、可打斷原理

4.1 不可打斷模式

在次模式下，即使它被打斷，仍會(huì)駐留在 AQS 隊(duì)列中，一直要等到獲得鎖后方能得知自己被打斷了。

我們都知道線程在知道自己沒有辦法獲取鎖的時(shí)候就會(huì)進(jìn)入到 acquireQueued() 方法的循環(huán)內(nèi)不斷的去嘗試，如果還不成功，就會(huì)進(jìn)入到parkAndCheckInterrupt() 方法里面等待。

但是進(jìn)入到 parkAndCheckInterrupt() 方法的線程可以被其他的線程調(diào)用它的 interrupted() 方法喚醒，如下面的parkAndCheckInterrupt() 方法，會(huì)返回一個(gè) boolean 表示是否被打斷過，但是interrupted() 方法還會(huì)清除打斷標(biāo)記，即下次再 park 還可以 park 住。

如果parkAndCheckInterrupt() 返回 true 了就會(huì)進(jìn)入 if 塊了，此時(shí)就會(huì)對(duì)打斷標(biāo)記做一個(gè)記錄，設(shè)置值為 true，沒有做額外的處理，接下來還是會(huì)繼續(xù) for 循環(huán)，獲取不了鎖還是會(huì) park 阻塞。只有當(dāng)你獲取到鎖以后，會(huì)把打斷標(biāo)記作為結(jié)果返回。

此時(shí)就會(huì)進(jìn)入到selfInterrupt() 方法，重新產(chǎn)生一次中斷。

static final class NonfairSync extends Sync {
    // ...

    private final boolean parkAndCheckInterrupt() {
        // 如果打斷標(biāo)記已經(jīng)是 true, 則 park 會(huì)失效
        LockSupport.park(this);
        // interrupted 會(huì)清除打斷標(biāo)記
        return Thread.interrupted();
    }

    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;
                    failed = false;
                    // 還是需要獲得鎖后, 才能返回打斷狀態(tài)
                    return interrupted;
                }
                if (
                        shouldParkAfterFailedAcquire(p, node) &&
                                parkAndCheckInterrupt()
                ) {
                    // 如果是因?yàn)?interrupt 被喚醒, 返回打斷狀態(tài)為 true
                    interrupted = true;
                }
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

    public final void acquire(int arg) {
        if (
                !tryAcquire(arg) &&
                        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
        ) {
            // 如果打斷狀態(tài)為 true
            selfInterrupt();
        }
    }

    static void selfInterrupt() {
        // 重新產(chǎn)生一次中斷
        Thread.currentThread().interrupt();
    }
}

4.2 可打斷模式

可打斷模式就是在調(diào)用acquireInterruptibly() 方法獲取鎖時(shí)可以被打斷，它和上面的方法區(qū)別是當(dāng)線程被喚醒的時(shí)候，即調(diào)用doAcquireInterruptibly() 時(shí)會(huì)拋出一個(gè)異常，打斷 for 循環(huán)等待，即線程停止去等待鎖

// Sync 繼承自 AQS
static final class NonfairSync extends Sync {
    public final void acquireInterruptibly(int arg) throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        // 如果沒有獲得到鎖, 進(jìn)入 ㈠
        if (!tryAcquire(arg))
            doAcquireInterruptibly(arg);
    }

    // ㈠ 可打斷的獲取鎖流程
    private void doAcquireInterruptibly(int arg) throws InterruptedException {
        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;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt()) {
                    // 在 park 過程中如果被 interrupt 會(huì)進(jìn)入此
                    // 這時(shí)候拋出異常, 而不會(huì)再次進(jìn)入 for (;;)
                    throw new InterruptedException();
                }
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
}

五、公平鎖實(shí)現(xiàn)原理

公平鎖和非公平鎖的區(qū)別主要就是在 tryAcquire() 方法的實(shí)現(xiàn)中，當(dāng) c==0 即還沒有別人占這個(gè)鎖時(shí)，先調(diào)用 hasQueuedPredecessors() 方法判斷隊(duì)列中是否有前驅(qū)節(jié)點(diǎn)，沒有才回去競(jìng)爭(zhēng)，即只能是老二節(jié)點(diǎn)線程才可以運(yùn)行，非老二節(jié)點(diǎn)線程不能運(yùn)行。

static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

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

    // AQS 繼承過來的方法, 方便閱讀, 放在此處
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
                        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
        ) {
            selfInterrupt();
        }
    }

    // 與非公平鎖主要區(qū)別在于 tryAcquire 方法的實(shí)現(xiàn)
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // 先檢查 AQS 隊(duì)列中是否有前驅(qū)節(jié)點(diǎn), 沒有才去競(jìng)爭(zhēng)
            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;
    }

    // ㈠ AQS 繼承過來的方法, 方便閱讀, 放在此處
    public final boolean hasQueuedPredecessors() {
        Node t = tail;
        Node h = head;
        Node s;
        // h != t 時(shí)表示隊(duì)列中有 Node
        return h != t &&
                (
                        // (s = h.next) == null 表示隊(duì)列中還有沒有老二
                        (s = h.next) == null ||
                                // 或者隊(duì)列中老二線程不是此線程
                                s.thread != Thread.currentThread()
                );
    }
}

六、條件變量實(shí)現(xiàn)原理

每個(gè)條件變量其實(shí)就對(duì)應(yīng)著一個(gè)等待隊(duì)列，其實(shí)現(xiàn)類就是 ConditionObject。

6.1 await 流程

假設(shè)一開始 Thread-0 持有鎖，如下圖所示：

接下來如果線程調(diào)用了ConditionObject 類的 await() 方法，它就會(huì)進(jìn)入到ConditionObject 的 addConditionWaiter 流程，如下圖：

addConditionWaiter () 方法會(huì)把調(diào)用的線程加入到條件變量里面的雙向鏈表之中，并且把新的 Node 狀態(tài)變?yōu)?strong> -2，表示在條件變量中處于等待狀態(tài)，如下圖：

接下來會(huì)進(jìn)入到AQS 的 fullyRelease 流程，釋放同步器上的鎖，如下圖：

接下來會(huì) unpark AQS 隊(duì)列中的下一個(gè)節(jié)點(diǎn)，競(jìng)爭(zhēng)鎖，假設(shè)沒有其他競(jìng)爭(zhēng)線程，那么 Thread-1 競(jìng)爭(zhēng)成功，如下圖所示：

此時(shí) Thread-0 的狀態(tài)就被 park 阻塞了，如下圖：

6.2 signal 流程

假設(shè)此時(shí) Thread-1 要來喚醒 Thread-0，如下圖：

signal() 方法的代碼如下所示，首先進(jìn)行健康檢查，看調(diào)用方法的線程是否為鎖的持有者，即只有 owner 才可以喚醒其他線程，接下來查找條件變量列表中的第一個(gè)元素，查找完畢之后調(diào)用 doSignal() 方法。

doSignal() 方法是取得等待隊(duì)列中第一個(gè) Node，即 Thread-0 所在 Node

此時(shí)的狀態(tài)圖如下所示：

獲取完Node 之后，調(diào)用 transferForSignal() 方法，將該 Node 加入 AQS 隊(duì)列尾部，將 Thread-0 的 waitStatus 改為 0，Thread-3 的 waitStatus 改為 -1

接下來就是 Thread-1 釋放鎖，把 state 置為 0，設(shè)置 owner 為 null，喚醒等待隊(duì)列中的下一個(gè)元素了，整個(gè)流程結(jié)束。

總結(jié)

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評(píng)論