ReentrantReadWriterLock源碼(state設計、讀寫鎖、共享鎖、獨佔鎖及鎖降級）

時間 2020-12-18

標籤 java node 編程 c# 併發 .net 線程設計 code 欄目 Java 简体版

原文原文鏈接

ReentrantReadWriterLock

讀寫鎖類圖(截圖來源https://blog.csdn.net/wangbo199308/article/details/108688148)
java

state的設計

讀寫鎖將變量state切分紅兩個部分，高16位表示讀，低16位表示寫node

源碼中將4字節(32位)的int數據類型state，經過SHARED_SHIFT(16)劃分讀和寫；編程

每次讀鎖增長的單元，SHARED_UNIT = (1 << SHARED_SHIFT) 也即0x00010000，即每次讀鎖增長從17位開始加1c#

讀寫鎖最大數量：MAX_COUNT = (1 << SHARED_SHIFT) - 1，16位最大值併發

寫鎖的掩碼：EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1，即求寫鎖數量，將state和此掩碼作與運算，將高16位抹去ui

計算讀鎖數量邏輯：c >>> SHARED_SHIFT，取高16位.net

計算寫鎖數量邏輯：c & EXCLUSIVE_MASK，將state和此掩碼作與運算，將高16位抹去線程

public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    abstract static class Sync extends AbstractQueuedSynchronizer {
        //16位劃分讀和寫
        static final int SHARED_SHIFT   = 16;
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
        
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
    }
}

讀鎖

讀鎖上鎖的調用鏈：ReentrantReadWriteLock$ReadLock#lock() -->AbstractQueuedSynchronizer#acquireShared() -->ReentrantReadWriteLock$Sync#tryAcquireShared()設計

當前寫鎖數量爲0或獨佔鎖持有者就是當前線程才進行讀鎖邏輯code

讀鎖數量經過CAS加1

以後邏輯是將讀鎖線程放入ThreadLocal中，記錄各自鎖數量

public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    public static class ReadLock implements Lock, java.io.Serializable {
        public void lock() {
            sync.acquireShared(1);
        }
    }
}

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
}

public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    abstract static class Sync extends AbstractQueuedSynchronizer {
        protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
            int c = getState();
            // 同時知足寫鎖數量不爲0，且獨佔鎖不是當前線程，走doAcquireShared邏輯
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
            // 取高16位讀鎖數量
            int r = sharedCount(c);
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                // ThreadLocal存放鎖信息
                if (r == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                } else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
            return fullTryAcquireShared(current);
        }
    }
}

在讀鎖獲取鎖過程，寫鎖不爲0且佔有寫鎖的不是當前線程，返回-1，走同步器doAcquireShared方法，等待寫鎖釋放；

前置節點是head節點時，嘗試獲取共享鎖

private void doAcquireShared(int arg) {
    // 隊列加入的node是共享模式
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                //前置節點是head節點時，嘗試獲取共享鎖
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

寫鎖

讀鎖不爲0，但寫鎖爲0，獲取鎖失敗；讀鎖不爲0，寫鎖也不爲0，但獨佔鎖不是當前線程，獲取鎖失敗
若是鎖數量已到最大，獲取失敗
不然獲取寫鎖，更新state

public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    abstract static class Sync extends AbstractQueuedSynchronizer {
        protected final boolean tryAcquire(int acquires) {

            Thread current = Thread.currentThread();
            int c = getState();
            int w = exclusiveCount(c);
            if (c != 0) {
                // (Note: if c != 0 and w == 0 then shared count != 0)
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // Reentrant acquire
                setState(c + acquires);
                return true;
            }
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            setExclusiveOwnerThread(current);
            return true;
        }
    }
}

共享鎖和獨佔鎖

讀鎖是共享鎖，當線程1得到讀鎖時，並不會排斥線程2去獲取讀鎖，而是在ThreadLocal中保存每一個鎖數量

abstract static class Sync extends AbstractQueuedSynchronizer {
        static final class HoldCounter {
            int count = 0;
            // Use id, not reference, to avoid garbage retention
            final long tid = getThreadId(Thread.currentThread());
        }
        
        static final class ThreadLocalHoldCounter
            extends ThreadLocal<HoldCounter> {
            public HoldCounter initialValue() {
                return new HoldCounter();
            }
        }
    }

寫鎖是獨佔鎖，會調用同步器AbstractQueuedSynchronizer#acquire()方法，默認加入隊列的node模式是獨佔模式

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

鎖降級

鎖降級就是從寫鎖降級成爲讀鎖。在當前線程擁有寫鎖的狀況下，再次獲取到讀鎖，隨後釋放寫鎖的過程就是鎖降級

鎖降級示例:

public void processData() {
    ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
    ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
    readLock.lock();
    if(!update) {
        //必須先釋放讀鎖
        readLock.unlock();
        // 鎖降級從寫鎖獲取到開始
        writeLock.lock();
        try{
            if(!update) {
                update = true;
            }
            // 能夠獲取到讀鎖，getExclusiveOwnerThread() == current
            readLock.lock();
        } finally {
            writeLock.unlock();
        }
        //鎖降級完成，寫鎖降級爲讀鎖
    }
    try{
        // 使用數據的流程
    } finally {
        readLock.unlock();
    }
}

可降級的源碼還是在讀鎖tryAcquireShared方法中，getExclusiveOwnerThread() == current，也即當前獨佔鎖owner就是當前線程，可進行讀鎖邏輯。

protected final int tryAcquireShared(int unused) {
    if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1;
}

參考：《Java併發編程的藝術》