天天看點

Java多線程進階—— J.U.C之locks架構:StampedLock

一、StampedLock類簡介

StampedLock類,在JDK1.8時引入,是對讀寫鎖ReentrantReadWriteLock的增強,該類提供了一些功能,優化了讀鎖、寫鎖的通路,同時使讀寫鎖之間可以互相轉換,更細粒度控制并發。

首先明确下,該類的設計初衷是作為一個内部工具類,用于輔助開發其它線程安全元件,用得好,該類可以提升系統性能,用不好,容易産生死鎖和其它莫名其妙的問題。

1.1 StampedLock的引入

先來看下,為什麼有了ReentrantReadWriteLock,還要引入StampedLock?

ReentrantReadWriteLock使得多個讀線程同時持有讀鎖(隻要寫鎖未被占用),而寫鎖是獨占的。

但是,讀寫鎖如果使用不當,很容易産生“饑餓”問題:

比如在讀線程非常多,寫線程很少的情況下,很容易導緻寫線程“饑餓”,雖然使用“公平”政策可以一定程度上緩解這個問題,但是“公平”政策是以犧牲系統吞吐量為代價的。(在ReentrantLock類的介紹章節中,介紹過這種情況)

1.2 StampedLock的特點

StampedLock的主要特點概括一下,有以下幾點:

  1. 所有擷取鎖的方法,都傳回一個郵戳(Stamp),Stamp為0表示擷取失敗,其餘都表示成功;
  2. 所有釋放鎖的方法,都需要一個郵戳(Stamp),這個Stamp必須是和成功擷取鎖時得到的Stamp一緻;
  3. StampedLock是不可重入的;(如果一個線程已經持有了寫鎖,再去擷取寫鎖的話就會造成死鎖)
  4. StampedLock有三種通路模式:

    ①Reading(讀模式):功能和ReentrantReadWriteLock的讀鎖類似

    ②Writing(寫模式):功能和ReentrantReadWriteLock的寫鎖類似

    ③Optimistic reading(樂觀讀模式):這是一種優化的讀模式。

  5. StampedLock支援讀鎖和寫鎖的互相轉換

    我們知道RRW中,當線程擷取到寫鎖後,可以降級為讀鎖,但是讀鎖是不能直接更新為寫鎖的。

    StampedLock提供了讀鎖和寫鎖互相轉換的功能,使得該類支援更多的應用場景。

  6. 無論寫鎖還是讀鎖,都不支援Conditon等待

我們知道,在ReentrantReadWriteLock中, 當讀鎖被使用時,如果有線程嘗試擷取寫鎖,該寫線程會阻塞。

但是,在Optimistic reading中,即使讀線程擷取到了讀鎖,寫線程嘗試擷取寫鎖也不會阻塞,這相當于對讀模式的優化,但是可能會導緻資料不一緻的問題。是以,當使用Optimistic reading擷取到讀鎖時,必須對擷取結果進行校驗。

二、StampedLock使用示例

先來看一個Oracle官方的例子:

class Point {
    private double x, y;
    private final StampedLock sl = new StampedLock();

    void move(double deltaX, double deltaY) {
        long stamp = sl.writeLock();    //涉及對共享資源的修改,使用寫鎖-獨占操作
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            sl.unlockWrite(stamp);
        }
    }

    /**
     * 使用樂觀讀鎖通路共享資源
     * 注意:樂觀讀鎖在保證資料一緻性上需要拷貝一份要操作的變量到方法棧,并且在操作資料時候可能其他寫線程已經修改了資料,
     * 而我們操作的是方法棧裡面的資料,也就是一個快照,是以最多傳回的不是最新的資料,但是一緻性還是得到保障的。
     *
     * @return
     */
    double distanceFromOrigin() {
        long stamp = sl.tryOptimisticRead();    // 使用樂觀讀鎖
        double currentX = x, currentY = y;      // 拷貝共享資源到本地方法棧中
        if (!sl.validate(stamp)) {              // 如果有寫鎖被占用,可能造成資料不一緻,是以要切換到普通讀鎖模式
            stamp = sl.readLock();             
            try {
                currentX = x;
                currentY = y;
            } finally {
                sl.unlockRead(stamp);
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }

    void moveIfAtOrigin(double newX, double newY) { // upgrade
        // Could instead start with optimistic, not read mode
        long stamp = sl.readLock();
        try {
            while (x == 0.0 && y == 0.0) {
                long ws = sl.tryConvertToWriteLock(stamp);  //讀鎖轉換為寫鎖
                if (ws != 0L) {
                    stamp = ws;
                    x = newX;
                    y = newY;
                    break;
                } else {
                    sl.unlockRead(stamp);
                    stamp = sl.writeLock();
                }
            }
        } finally {
            sl.unlock(stamp);
        }
    }
}
           

可以看到,上述示例最特殊的其實是distanceFromOrigin方法,這個方法中使用了“Optimistic reading”樂觀讀鎖,使得讀寫可以并發執行,但是“Optimistic reading”的使用必須遵循以下模式:

long stamp = lock.tryOptimisticRead();  // 非阻塞擷取版本資訊
copyVaraibale2ThreadMemory();           // 拷貝變量到線程本地堆棧
if(!lock.validate(stamp)){              // 校驗
    long stamp = lock.readLock();       // 擷取讀鎖
    try {
        copyVaraibale2ThreadMemory();   // 拷貝變量到線程本地堆棧
     } finally {
       lock.unlock(stamp);              // 釋放悲觀鎖
    }

}
useThreadMemoryVarables();              // 使用線程本地堆棧裡面的資料進行操作
           

三、StampedLock原理

3.1 StampedLock的内部常量

StampedLock雖然不像其它鎖一樣定義了内部類來實作AQS架構,但是StampedLock的基本實作思路還是利用CLH隊列進行線程的管理,通過同步狀态值來表示鎖的狀态和類型。

StampedLock内部定義了很多常量,定義這些常量的根本目的還是和ReentrantReadWriteLock一樣,對同步狀态值按位切分,以通過位運算對State進行操作:

對于StampedLock來說,寫鎖被占用的标志是第8位為1,讀鎖使用0-7位,正常情況下讀鎖數目為1-126,超過126時,使用一個名為

readerOverflow

的int整型儲存超出數。
Java多線程進階—— J.U.C之locks架構:StampedLock

部分常量的比特位表示如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

另外,StampedLock相比ReentrantReadWriteLock,對多核CPU進行了優化,可以看到,當CPU核數超過1時,會有一些自旋操作:

Java多線程進階—— J.U.C之locks架構:StampedLock

3.2 示例分析

假設現在有五個線程:ThreadA、ThreadB、ThreadC、ThreadD、ThreadE。操作如下:

//ThreadA調用writeLock, 擷取寫鎖

//ThreadB調用readLock, 擷取讀鎖

//ThreadC調用readLock, 擷取讀鎖

//ThreadD調用writeLock, 擷取寫鎖

//ThreadE調用readLock, 擷取讀鎖

1. StampedLock對象的建立

StampedLock的構造器很簡單,構造時設定下同步狀态值:

Java多線程進階—— J.U.C之locks架構:StampedLock

另外,StamedLock提供了三類視圖:

Java多線程進階—— J.U.C之locks架構:StampedLock

這些視圖其實是對StamedLock方法的封裝,便于習慣了ReentrantReadWriteLock的使用者使用:

例如,ReadLockView其實相當于

ReentrantReadWriteLock.readLock()

傳回的讀鎖;

Java多線程進階—— J.U.C之locks架構:StampedLock

2. ThreadA調用writeLock擷取寫鎖

來看下writeLock方法:

Java多線程進階—— J.U.C之locks架構:StampedLock

StampedLock中大量運用了位運算,這裡

(s = state) & ABITS == 0L

 表示讀鎖和寫鎖都未被使用,這裡寫鎖可以立即擷取成功,然後CAS操作更新同步狀态值State。

操作完成後,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

注意:StampedLock中,等待隊列的結點要比AQS中簡單些,僅僅三種狀态。

0:初始狀态

-1:等待中

1:取消

另外,結點的定義中有個

cowait

字段,該字段指向一個棧,用于儲存讀線程,這個後續會講到。

Java多線程進階—— J.U.C之locks架構:StampedLock

3. ThreadB調用readLock擷取讀鎖

來看下readLock方法:

由于ThreadA此時持有寫鎖,是以ThreadB擷取讀鎖失敗,将調用acquireRead方法,加入等待隊列:

Java多線程進階—— J.U.C之locks架構:StampedLock

acquireRead方法非常複雜,用到了大量自旋操作:

/**
 * 嘗試自旋的擷取讀鎖, 擷取不到則加入等待隊列, 并阻塞線程
 *
 * @param interruptible true 表示檢測中斷, 如果線程被中斷過, 則最終傳回INTERRUPTED
 * @param deadline      如果非0, 則表示限時擷取
 * @return 非0表示擷取成功, INTERRUPTED表示中途被中斷過
 */
private long acquireRead(boolean interruptible, long deadline) {
    WNode node = null, p;   // node指向入隊結點, p指向入隊前的隊尾結點

    /**
     * 自旋入隊操作
     * 如果寫鎖未被占用, 則立即嘗試擷取讀鎖, 擷取成功則傳回.
     * 如果寫鎖被占用, 則将目前讀線程包裝成結點, 并插入等待隊列(如果隊尾是寫結點,直接連結到隊尾;否則,連結到隊尾讀結點的棧中)
     */
    for (int spins = -1; ; ) {
        WNode h;
        if ((h = whead) == (p = wtail)) {   // 如果隊列為空或隻有頭結點, 則會立即嘗試擷取讀鎖
            for (long m, s, ns; ; ) {
                if ((m = (s = state) & ABITS) < RFULL ?     // 判斷寫鎖是否被占用
                    U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :  //寫鎖未占用,且讀鎖數量未超限, 則更新同步狀态
                    (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L))        //寫鎖未占用,但讀鎖數量超限, 超出部分放到readerOverflow字段中
                    return ns;          // 擷取成功後, 直接傳回
                else if (m >= WBIT) {   // 寫鎖被占用,以随機方式探測是否要退出自旋
                    if (spins > 0) {
                        if (LockSupport.nextSecondarySeed() >= 0)
                            --spins;
                    } else {
                        if (spins == 0) {
                            WNode nh = whead, np = wtail;
                            if ((nh == h && np == p) || (h = nh) != (p = np))
                                break;
                        }
                        spins = SPINS;
                    }
                }
            }
        }
        if (p == null) {                            // p == null表示隊列為空, 則初始化隊列(構造頭結點)
            WNode hd = new WNode(WMODE, null);
            if (U.compareAndSwapObject(this, WHEAD, null, hd))
                wtail = hd;
        } else if (node == null) {                  // 将目前線程包裝成讀結點
            node = new WNode(RMODE, p);
        } else if (h == p || p.mode != RMODE) {     // 如果隊列隻有一個頭結點, 或隊尾結點不是讀結點, 則直接将結點連結到隊尾, 連結完成後退出自旋
            if (node.prev != p)
                node.prev = p;
            else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
                p.next = node;
                break;
            }
        }
        // 隊列不為空, 且隊尾是讀結點, 則将添加目前結點連結到隊尾結點的cowait鍊中(實際上構成一個棧, p是棧頂指針 )
        else if (!U.compareAndSwapObject(p, WCOWAIT, node.cowait = p.cowait, node)) {    // CAS操作隊尾結點p的cowait字段,實際上就是頭插法插入結點
            node.cowait = null;
        } else {
            for (; ; ) {
                WNode pp, c;
                Thread w;
                // 嘗試喚醒頭結點的cowait中的第一個元素, 假如是讀鎖會通過循環釋放cowait鍊
                if ((h = whead) != null && (c = h.cowait) != null &&
                    U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
                    (w = c.thread) != null) // help release
                    U.unpark(w);
                if (h == (pp = p.prev) || h == p || pp == null) {
                    long m, s, ns;
                    do {
                        if ((m = (s = state) & ABITS) < RFULL ?
                            U.compareAndSwapLong(this, STATE, s,
                                ns = s + RUNIT) :
                            (m < WBIT &&
                                (ns = tryIncReaderOverflow(s)) != 0L))
                            return ns;
                    } while (m < WBIT);
                }
                if (whead == h && p.prev == pp) {
                    long time;
                    if (pp == null || h == p || p.status > 0) {
                        node = null; // throw away
                        break;
                    }
                    if (deadline == 0L)
                        time = 0L;
                    else if ((time = deadline - System.nanoTime()) <= 0L)
                        return cancelWaiter(node, p, false);
                    Thread wt = Thread.currentThread();
                    U.putObject(wt, PARKBLOCKER, this);
                    node.thread = wt;
                    if ((h != pp || (state & ABITS) == WBIT) && whead == h && p.prev == pp) {
                        // 寫鎖被占用, 且目前結點不是隊首結點, 則阻塞目前線程
                        U.park(false, time);
                    }
                    node.thread = null;
                    U.putObject(wt, PARKBLOCKER, null);
                    if (interruptible && Thread.interrupted())
                        return cancelWaiter(node, p, true);
                }
            }
        }
    }

    for (int spins = -1; ; ) {
        WNode h, np, pp;
        int ps;
        if ((h = whead) == p) {     // 如果目前線程是隊首結點, 則嘗試擷取讀鎖
            if (spins < 0)
                spins = HEAD_SPINS;
            else if (spins < MAX_HEAD_SPINS)
                spins <<= 1;
            for (int k = spins; ; ) { // spin at head
                long m, s, ns;
                if ((m = (s = state) & ABITS) < RFULL ?     // 判斷寫鎖是否被占用
                    U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :  //寫鎖未占用,且讀鎖數量未超限, 則更新同步狀态
                    (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L)) {      //寫鎖未占用,但讀鎖數量超限, 超出部分放到readerOverflow字段中
                    // 擷取讀鎖成功, 釋放cowait鍊中的所有讀結點
                    WNode c;
                    Thread w;

                    // 釋放頭結點, 目前隊首結點成為新的頭結點
                    whead = node;
                    node.prev = null;

                    // 從棧頂開始(node.cowait指向的結點), 依次喚醒所有讀結點, 最終node.cowait==null, node成為新的頭結點
                    while ((c = node.cowait) != null) {
                        if (U.compareAndSwapObject(node, WCOWAIT, c, c.cowait) && (w = c.thread) != null)
                            U.unpark(w);
                    }
                    return ns;
                } else if (m >= WBIT &&
                    LockSupport.nextSecondarySeed() >= 0 && --k <= 0)
                    break;
            }
        } else if (h != null) {     // 如果頭結點存在cowait鍊, 則喚醒鍊中所有讀線程
            WNode c;
            Thread w;
            while ((c = h.cowait) != null) {
                if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
                    (w = c.thread) != null)
                    U.unpark(w);
            }
        }
        if (whead == h) {
            if ((np = node.prev) != p) {
                if (np != null)
                    (p = np).next = node;   // stale
            } else if ((ps = p.status) == 0)        // 将前驅結點的等待狀态置為WAITING, 表示之後将喚醒目前結點
                U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
            else if (ps == CANCELLED) {
                if ((pp = p.prev) != null) {
                    node.prev = pp;
                    pp.next = node;
                }
            } else {        // 阻塞目前讀線程
                long time;
                if (deadline == 0L)
                    time = 0L;
                else if ((time = deadline - System.nanoTime()) <= 0L)   //限時等待逾時, 取消等待
                    return cancelWaiter(node, node, false);

                Thread wt = Thread.currentThread();
                U.putObject(wt, PARKBLOCKER, this);
                node.thread = wt;
                if (p.status < 0 && (p != h || (state & ABITS) == WBIT) && whead == h && node.prev == p) {
                    // 如果前驅的等待狀态為WAITING, 且寫鎖被占用, 則阻塞目前調用線程
                    U.park(false, time);
                }
                node.thread = null;
                U.putObject(wt, PARKBLOCKER, null);
                if (interruptible && Thread.interrupted())
                    return cancelWaiter(node, node, true);
            }
        }
    }
}
           

我們來分析下這個方法。

該方法會首先自旋的嘗試擷取讀鎖,擷取成功後,就直接傳回;否則,會将目前線程包裝成一個讀結點,插入到等待隊列。

由于,目前等待隊列還是空,是以ThreadB會初始化隊列,然後将自身包裝成一個讀結點,插入隊尾,然後在下面這個地方跳出自旋:

Java多線程進階—— J.U.C之locks架構:StampedLock

此時,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

跳出自旋後,ThreadB會繼續向下執行,進入下一個自旋,在下一個自旋中,依然會再次嘗試擷取讀鎖,如果這次再擷取不到,就會将前驅的等待狀态置為WAITING, 表示我(目前線程)要去睡了(阻塞),到時記得叫醒我:

Java多線程進階—— J.U.C之locks架構:StampedLock
Java多線程進階—— J.U.C之locks架構:StampedLock

最終, ThreadB進入阻塞狀态:

Java多線程進階—— J.U.C之locks架構:StampedLock

最終,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

4. ThreadC調用readLock擷取讀鎖

這個過程和ThreadB擷取讀鎖一樣,差別在于ThreadC被包裝成結點加入等待隊列後,是連結到ThreadB結點的棧指針中的。調用完下面這段代碼後,ThreadC會連結到以Thread B為棧頂指針的棧中:

Java多線程進階—— J.U.C之locks架構:StampedLock
Java多線程進階—— J.U.C之locks架構:StampedLock
注意:讀結點的cowait字段其實構成了一個棧,入棧的過程其實是個“頭插法”插入單連結清單的過程。比如,再來個ThreadX讀結點,則cowait連結清單結構為:

ThreadB - > ThreadX -> ThreadC

。最終喚醒讀結點時,将從棧頂開始。

然後會在下一次自旋中,阻塞目前讀線程:

Java多線程進階—— J.U.C之locks架構:StampedLock

最終,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

可以看到,此時ThreadC結點并沒有把它的前驅的等待狀态置為-1,因為ThreadC是連結到棧中的,當寫鎖釋放的時候,會從棧底元素開始,喚醒棧中所有讀結點。

5. ThreadD調用writeLock擷取寫鎖

ThreadD調用writeLock方法擷取寫鎖失敗後(ThreadA依然占用着寫鎖),會調用acquireWrite方法,該方法整體邏輯和acquireRead差不多,首先自旋的嘗試擷取寫鎖,擷取成功後,就直接傳回;否則,會将目前線程包裝成一個寫結點,插入到等待隊列。

Java多線程進階—— J.U.C之locks架構:StampedLock

acquireWrite源碼:

/**
 * 嘗試自旋的擷取寫鎖, 擷取不到則阻塞線程
 *
 * @param interruptible true 表示檢測中斷, 如果線程被中斷過, 則最終傳回INTERRUPTED
 * @param deadline      如果非0, 則表示限時擷取
 * @return 非0表示擷取成功, INTERRUPTED表示中途被中斷過
 */
private long acquireWrite(boolean interruptible, long deadline) {
    WNode node = null, p;

    /**
     * 自旋入隊操作
     * 如果沒有任何鎖被占用, 則立即嘗試擷取寫鎖, 擷取成功則傳回.
     * 如果存在鎖被使用, 則将目前線程包裝成獨占結點, 并插入等待隊列尾部
     */
    for (int spins = -1; ; ) {
        long m, s, ns;
        if ((m = (s = state) & ABITS) == 0L) {      // 沒有任何鎖被占用
            if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT))    // 嘗試立即擷取寫鎖
                return ns;                                                 // 擷取成功直接傳回
        } else if (spins < 0)
            spins = (m == WBIT && wtail == whead) ? SPINS : 0;
        else if (spins > 0) {
            if (LockSupport.nextSecondarySeed() >= 0)
                --spins;
        } else if ((p = wtail) == null) {       // 隊列為空, 則初始化隊列, 構造隊列的頭結點
            WNode hd = new WNode(WMODE, null);
            if (U.compareAndSwapObject(this, WHEAD, null, hd))
                wtail = hd;
        } else if (node == null)               // 将目前線程包裝成寫結點
            node = new WNode(WMODE, p);
        else if (node.prev != p)
            node.prev = p;
        else if (U.compareAndSwapObject(this, WTAIL, p, node)) {    // 連結結點至隊尾
            p.next = node;
            break;
        }
    }

    for (int spins = -1; ; ) {
        WNode h, np, pp;
        int ps;
        if ((h = whead) == p) {     // 如果目前結點是隊首結點, 則立即嘗試擷取寫鎖
            if (spins < 0)
                spins = HEAD_SPINS;
            else if (spins < MAX_HEAD_SPINS)
                spins <<= 1;
            for (int k = spins; ; ) { // spin at head
                long s, ns;
                if (((s = state) & ABITS) == 0L) {      // 寫鎖未被占用
                    if (U.compareAndSwapLong(this, STATE, s,
                        ns = s + WBIT)) {               // CAS修改State: 占用寫鎖
                        // 将隊首結點從隊列移除
                        whead = node;
                        node.prev = null;
                        return ns;
                    }
                } else if (LockSupport.nextSecondarySeed() >= 0 &&
                    --k <= 0)
                    break;
            }
        } else if (h != null) {  // 喚醒頭結點的棧中的所有讀線程
            WNode c;
            Thread w;
            while ((c = h.cowait) != null) {
                if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) && (w = c.thread) != null)
                    U.unpark(w);
            }
        }
        if (whead == h) {
            if ((np = node.prev) != p) {
                if (np != null)
                    (p = np).next = node;   // stale
            } else if ((ps = p.status) == 0)        // 将目前結點的前驅置為WAITING, 表示目前結點會進入阻塞, 前驅将來需要喚醒我
                U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
            else if (ps == CANCELLED) {
                if ((pp = p.prev) != null) {
                    node.prev = pp;
                    pp.next = node;
                }
            } else {        // 阻塞目前調用線程
                long time;  // 0 argument to park means no timeout
                if (deadline == 0L)
                    time = 0L;
                else if ((time = deadline - System.nanoTime()) <= 0L)
                    return cancelWaiter(node, node, false);
                Thread wt = Thread.currentThread();
                U.putObject(wt, PARKBLOCKER, this);
                node.thread = wt;
                if (p.status < 0 && (p != h || (state & ABITS) != 0L) && whead == h && node.prev == p)
                    U.park(false, time);    // emulate LockSupport.park
                node.thread = null;
                U.putObject(wt, PARKBLOCKER, null);
                if (interruptible && Thread.interrupted())
                    return cancelWaiter(node, node, true);
            }
        }
    }
}
           

acquireWrite中的下面這個自旋操作,用于将線程包裝成寫結點,插入隊尾:

Java多線程進階—— J.U.C之locks架構:StampedLock

插入完成後,隊列結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

然後,進入下一個自旋,并在下一個自旋中阻塞ThreadD,最終隊列結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

6. ThreadE調用readLock擷取讀鎖

同樣,由于寫鎖被ThreadA占用着,是以最終會調用acquireRead方法,在該方法的第一個自旋中,會将ThreadE加入等待隊列:

Java多線程進階—— J.U.C之locks架構:StampedLock
注意,由于隊尾結點是寫結點,是以目前讀結點會直接連結到隊尾;如果隊尾是讀結點,則會連結到隊尾讀結點的cowait鍊中。

然後進入第二個自旋,阻塞ThreadE,最終隊列結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

7. ThreadA調用unlockWrite釋放寫鎖

通過CAS操作,修改State成功後,會調用release方法喚醒等待隊列的隊首結點:

Java多線程進階—— J.U.C之locks架構:StampedLock

release方法非常簡單,先将頭結點的等待狀态置為0,表示即将喚醒後繼結點,然後立即喚醒隊首結點:

Java多線程進階—— J.U.C之locks架構:StampedLock

此時,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

8. ThreadB被喚醒後繼續向下執行

ThreadB被喚醒後,會從原阻塞處繼續向下執行,然後開始下一次自旋:

Java多線程進階—— J.U.C之locks架構:StampedLock

第二次自旋時,ThreadB發現寫鎖未被占用,則成功擷取到讀鎖,然後從棧頂(ThreadB的cowait指針指向的結點)開始喚醒棧中所有線程,

最後傳回:

Java多線程進階—— J.U.C之locks架構:StampedLock

最終,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

9. ThreadC被喚醒後繼續向下執行

ThreadC被喚醒後,繼續執行,并進入下一次自旋,下一次自旋時,會成功擷取到讀鎖。

Java多線程進階—— J.U.C之locks架構:StampedLock

注意,此時ThreadB和ThreadC已經拿到了讀鎖,ThreadD(寫線程)和ThreadE(讀線程)依然阻塞中,原來ThreadC對應的結點是個孤立結點,會被GC回收。

最終,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

10. ThreadB和ThreadC釋放讀鎖

ThreadB和ThreadC調用unlockRead方法釋放讀鎖,CAS操作State将讀鎖數量減1:

Java多線程進階—— J.U.C之locks架構:StampedLock

注意,當讀鎖的數量變為0時才會調用release方法,喚醒隊首結點:

Java多線程進階—— J.U.C之locks架構:StampedLock

隊首結點(ThreadD寫結點被喚醒),最終等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

11. ThreadD被喚醒後繼續向下執行

ThreadD會從原阻塞處繼續向下執行,并在下一次自旋中擷取到寫鎖,然後傳回:

Java多線程進階—— J.U.C之locks架構:StampedLock

最終,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

12. ThreadD調用unlockWrite釋放寫鎖

ThreadD釋放寫鎖的過程和步驟7完全相同,會調用unlockWrite喚醒隊首結點(ThreadE)。

Java多線程進階—— J.U.C之locks架構:StampedLock

ThreadE被喚醒後會從原阻塞處繼續向下執行,但由于ThreadE是個讀結點,是以同時會喚醒cowait棧中的所有讀結點,過程和 步驟8 完全一樣。最終,等待隊列的結構如下:

Java多線程進階—— J.U.C之locks架構:StampedLock

至此,全部執行完成。

四、StampedLock類/方法聲明

參考Oracle官方文檔:https://docs.oracle.com/javas...

類聲明:

Java多線程進階—— J.U.C之locks架構:StampedLock

方法聲明:

Java多線程進階—— J.U.C之locks架構:StampedLock

五、StampedLock總結

StampedLock的等待隊列與RRW的CLH隊列相比,有以下特點:

  1. 當入隊一個線程時,如果隊尾是讀結點,不會直接連結到隊尾,而是連結到該讀結點的cowait鍊中,cowait鍊本質是一個棧;
  2. 當入隊一個線程時,如果隊尾是寫結點,則直接連結到隊尾;
  3. 喚醒線程的規則和AQS類似,都是首先喚醒隊首結點。差別是StampedLock中,當喚醒的結點是讀結點時,會喚醒該讀結點的cowait鍊中的所有讀結點(順序和入棧順序相反,也就是後進先出)。

另外,StampedLock使用時要特别小心,避免鎖重入的操作,在使用樂觀讀鎖時也需要遵循相應的調用模闆,防止出現資料不一緻的問題。