Java多线程篇(9)——AQS之读写锁（ReentrantReadWriteLock）

1、读写锁的实现

1.1、state的分割与HoldCounter

ReentrantReadWriteLock 内部维护了读锁和写锁两个锁，这两个锁内部都依赖于同一个sync变量，也即使用了同一个aqs队列。

写锁：其实就是独占锁的获取释放

读锁：其实就是共享锁的获取释放

所以一个sync是如何既实现共享锁又实现独占锁的？换句话说就是一个 state 变量是如何既表示共享锁又表示独占锁的？
答：采用“按位切割使用”的方式来维护这个变量，将其切分为两部分：高16用于共享锁，低16用于独占锁。
写状态，等于 state & 0x0000FFFF（将高 16 位全部抹去）。 当写状态加1，等于 state + 1。
读状态，等于 state >>> 16 (无符号补 0 右移 16 位)。当读状态加1，等于 state +（1<<16）,也就是 state + 0x00010000。
获取读锁之前先判断是否有写锁（低16位是否为0），如果没有写锁才可以获取读锁。同理，获取写锁之前先判断是否有读锁或写锁（state是否为0），如果没有任何锁才可以获取写锁。

读写锁：读写，写写互斥，读读不互斥

exclusiveCount：低16位，写锁（重入）次数。
sharedCount：高16位，读锁的线程数。

在看实际的读写逻辑之前，先思考这么一个问题。
因为读锁是多个线程持有的，所以不能像写锁一样重入次数直接记录在state，那么读锁的重入次数记录在哪？
答：读锁的重入次数记录在 ThreadLocal，每个线程都维护了一个 HoldCounter 计数器。除第一个读锁的线程是直接记录在 firstReaderHoldCount 变量外，其他线程的重入次数都记录在 ThreadLocal。

1.2、写锁的获取/释放

Sync.tryAcquire

		protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();
            
            //得到 state 以及位运算得到 state 的低16位
            int c = getState();
            int w = exclusiveCount(c);

			//如果state不为0则只需判断是否写锁重入的情况
            if (c != 0) {
            	//如果低16位为0，或者当前线程不是独占线程都不是写锁重入的情况
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                //是写锁重入的情况，state+acquires
                setState(c + acquires);
                return true;
            }
            
            //如果state为0，说明是无锁状态，直接cas(state+acquires)。
            //其中writerShouldBlock()在公平/非公平锁有不同的实现
            //  非公平锁：直接返回false
            //	公平锁：如果队列有节点返回true，反之false
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
                
            //设置当前独占锁的线程
            setExclusiveOwnerThread(current);
            return true;
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

Sync.tryRelease

        protected final boolean tryRelease(int releases) {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            //就是直接的state-releases，然后设值
            int nextc = getState() - releases;
            boolean free = exclusiveCount(nextc) == 0;
            if (free)
            	//如果重入释放完了，就清空独占锁线程
                setExclusiveOwnerThread(null);
            setState(nextc);
            return free;
        }
1
2
3
4
5
6
7
8
9
10
11
12

1.3、读锁的获取/释放

Sync.tryAcquireShared

		protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
            int c = getState();
            
            //用于写锁降级的情况
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
                
            //获取state的高16位
            int r = sharedCount(c);
            
            //cas(state+0x00010000)
            //其中readerShouldBlock() 在公平/非公平锁有不同的实现
            // 非公平锁：理论上也是直接返回false，但是为了避免大量的读线程导致写线程的饥饿。
            //			所以这里做了简单的优化：如果队列中第一个等待的线程是写线程，就返回true
            // 公平锁：如果队列有节点返回true，反之false
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                //如果是第一个线程获取的读锁，就将重入次数记入firstReaderHoldCount 
                if (r == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                }
                //反之记录在 HoldCounter 
                else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null ||
                        rh.tid != LockSupport.getThreadId(current))
                        //readHolds是ThreadLocal的子类
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
            
            // 上面获锁失败，这里通过自旋的方式再次尝试获取读锁，逻辑跟上面的类似。
            return fullTryAcquireShared(current);
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

Sync.tryReleaseShared

        protected final boolean tryReleaseShared(int unused) {
            Thread current = Thread.currentThread();
            
            //如果是第一个获取读锁的线程 firstReaderHoldCount--
            if (firstReader == current) {
                if (firstReaderHoldCount == 1)
                    firstReader = null;
                else
                    firstReaderHoldCount--;
            } 
            //反之 HoldCounter--
            else {
                HoldCounter rh = cachedHoldCounter;
                if (rh == null ||
                    rh.tid != LockSupport.getThreadId(current))
                    //readHolds是ThreadLocal的子类
                    rh = readHolds.get();
                int count = rh.count;
                if (count <= 1) {
                    readHolds.remove();
                    if (count <= 0)
                        throw unmatchedUnlockException();
                }
                --rh.count;
            }
            
            //cas(state- 0x00010000)
            for (;;) {
                int c = getState();
                int nextc = c - SHARED_UNIT;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

综上所述，内部设计如下

2、写锁降级成读锁的使用场景

在上面读锁获取的代码中有这么一段

		protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
            int c = getState();
            
            //用于写锁降级的情况
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;

			//...
        }
1
2
3
4
5
6
7
8
9
10
11

什么场景下写需要锁的降级？

比如这种先写后读的场景，即写完之后，后续都是读操作。后续读操作的独占锁是没必要的，为提高吞吐量可以把独占锁换成共享锁。
那为什么不是：写锁->写操作->释放写锁->读锁->读操作->释放读锁？即为什么获取读锁要在释放写锁之前？
因为如果先释放写锁再获取读锁，从释放写锁到获取读锁中间可能有其他线程先获取到了写锁并对修改了值，此时对于原来的线程来说就产生了数据前后不一致的问题。（前后两行代码，释放写锁时明明还是好好的，获取读锁的时候就变了）