ReentrantReadWriteLock 源码分析以及 AQS 共享锁 （二）
- 2020 年 3 月 17 日
- 筆記
前言
上一篇讲解了 AQS 的独占锁部分（参看：ReentrantLock 源码分析以及 AQS （一）），这一篇将介绍 AQS 的共享锁，以及基于共享锁实现读写锁分离的 ReentrantReadWriteLock。（若是遇到之前讲过的方法，将不再赘述）
先思考一下，为什么我们用读写锁分离？
我们知道 ReentrantLock 用的是独占锁，不管线程是读还是写状态，都会阻塞，这无疑会降低并发量。
但是，我们知道多个线程同时去读数据的时候，并不会产生线程安全的问题，因为它们互不干扰。那么为什么不设计一种方案，让所有的读线程可以共享，一起同时读数据呢，只需要阻塞写的线程就可以了。提高并发的同时，也不会产生数据不一致的现象。
同样的，如果有线程在写数据，那么也会阻塞其它读线程（同样阻塞其它写线程），数据写完之后才可以读数据，这样保证读到的数据都是最新的。
因此，我们可以用读、写两把锁，分别控制数据的读和写。实现读读共享、读写互斥，写写互斥。这也是 ReentrantReadWriteLock 读写分离锁的由来。它非常适合用在读多写少的场景。
ReentrantReadWriteLock
它和 ReentrantLock 一样，也是一个可重入的锁，并基于 AQS 共享锁实现了读写分离。其内部结构也大同小异，支持公平锁和非公平锁。我们看下它的构造函数，
public ReentrantReadWriteLock() { //默认非公平 this(false); } public ReentrantReadWriteLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); readerLock = new ReadLock(this); writerLock = new WriteLock(this); }它定义了两个内部类来表示读锁和写锁，并且都通过内部类 Sync 来实现加锁，释放锁等功能。
public static class ReadLock implements Lock, java.io.Serializable { private static final long serialVersionUID = -5992448646407690164L; private final Sync sync; protected ReadLock(ReentrantReadWriteLock lock) { sync = lock.sync; } ... } public static class WriteLock implements Lock, java.io.Serializable { private static final long serialVersionUID = -4992448646407690164L; private final Sync sync; protected WriteLock(ReentrantReadWriteLock lock) { sync = lock.sync; } ... } abstract static class Sync extends AbstractQueuedSynchronizer { }我们再看下公平锁和非公平锁，其中有两个比较重要的方法，用来判断读锁和写锁是否应该被阻塞，后面加锁的时候会用到（其实，实际情况是否真的应该阻塞，还需要斟酌，后面会说）。
static final class FairSync extends Sync { private static final long serialVersionUID = -2274990926593161451L; //公平锁的读和写都需要判断，在它前面是否已经有线程在等待。 //有的话，当前线程就需要阻塞，这也体现了公平性。 final boolean writerShouldBlock() { return hasQueuedPredecessors(); } final boolean readerShouldBlock() { return hasQueuedPredecessors(); } } static final class NonfairSync extends Sync { private static final long serialVersionUID = -8159625535654395037L; //非公平锁，写的时候不需要阻塞，直接返回false final boolean writerShouldBlock() { return false; // writers can always barge } final boolean readerShouldBlock() { //为了避免写线程饥饿，需要判断同步队列中第一个排队的(head.next)是否是独占锁(写线程) //如果是的话，当前读线程就需要阻塞，这是 AQS 中的方法 return apparentlyFirstQueuedIsExclusive(); } } final boolean apparentlyFirstQueuedIsExclusive() { Node h, s; return (h = head) != null && (s = h.next) != null && !s.isShared() && s.thread != null; }思考：
我们知道 ReentrantLock 的同步状态和重入次数，是直接用 state 值来表示的。那么，现在我需要读和写两把锁，怎么才能用一个 int 类型的值来表示两把锁的状态呢？并且，锁是可重入的，重入的次数怎么记录呢？
别急，下面一个一个说。
怎么用一个 state 值表示读、写两把锁？
state 是一个 32 位的 int 值，读写锁中，把它一分为二，高 16 位用来表示读状态，其值代表读锁的线程数，如图中为 3 个，低 16位表示写状态，其值代表写锁的重入次数（因为是独占锁）。 这样，就可以分别计算读锁和写锁的个数了。其相关的属性和方法定义在 Sync 类中。
static final int SHARED_SHIFT = 16; //表明读锁每增加一个，state的实际值增加 2^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; //持有读锁的线程个数，参数如的 c 代表 state值 //state 的32位二进制位，无符号右移 16位之后，其实就是高16位的值 static int sharedCount(int c) { return c >>> SHARED_SHIFT; } //写锁数量，即写锁的重入次数 static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }读锁的个数计算比较简单，直接无符号右移 16 位即可。我们看下写锁的重入次数是怎么计算的。先看下 EXCLUSIVE_MASK 这个值，是 (1 << 16) – 1，我们用二进制表示计算过程为：
// 1的二进制 0000 0000 0000 0000 0000 0000 0000 0001 // 1左移 16位 0000 0000 0000 0001 0000 0000 0000 0000 //再减 1 0000 0000 0000 0000 1111 1111 1111 1111 //任何一个 32位二进制数 c，和以上值做 “与” 运算都为它本身 c 的低 16 位值 //这个不用解释了吧，这个不会的话，需要好好补充一下基础知识了。。。锁的重入次数是怎么计算的？
写锁比较简单，直接用计算出来的低16位值就可以代表写锁的重入次数。
读锁，就比较复杂了，因为高16位只能表示持有共享锁的线程个数，实在是分身乏术啊。所以，在 Sync 内部，维护了一个类，用来表示每个线程重入的次数，
static final class HoldCounter { int count = 0; // Use id, not reference, to avoid garbage retention final long tid = getThreadId(Thread.currentThread()); }这里边定义了一个计数器来表示重入次数，tid 来表示当前的线程 id 。但是，这样还不够，我们需要把 HoldCounter 和 线程绑定，这样才可以区分出来每个线程分别持有的锁个数（重入次数），这就需要用到 ThreadLocal 了。
static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> { //重写此方法，可以在 ThreadLocal 没有当前线程计数的情况下， //直接使用 的 get 方法，初始化一个，而不必 new 一个对象 public HoldCounter initialValue() { return new HoldCounter(); } }除此之外，Sync 中还定义了一些其他和读锁相关的属性，
//保存了当前线程重入的读锁次数，当重入次数减到 0 时移除 //移除应该是为了性能着想，因为可以随时通过 get 方法初始化 HoldCounter private transient ThreadLocalHoldCounter readHolds; //保存了最近一个获取读锁成功的线程计数，这个变量的目的是： //如果最后一个获取到读锁的线程重复获取读锁，那么就可以直接拿来用，而不用更新。 //相当于缓存，提高效率 private transient HoldCounter cachedHoldCounter; //第一个获取读锁的线程 private transient Thread firstReader = null; //第一个获取读锁的线程计数 private transient int firstReaderHoldCount; //这两个参数，是为了效率问题，当只有一个线程获得读锁时，就避免了查找 readHolds基本知识讲完啦，那么接下来就是锁的获取和释放了。先说下写锁吧，因为有上一篇独占锁的基础了，理解起来比较容易。
写锁的获取
写锁的获取从 lock 方法开始，
//ReentrantReadWriteLock.WriteLock.lock public void lock() { sync.acquire(1); } //AbstractQueuedSynchronizer.acquire public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } //公平锁和非公平锁调用的是同一个方法，在 Sync 类中定义 //ReentrantReadWriteLock.Sync.tryAcquire protected final boolean tryAcquire(int acquires) { Thread current = Thread.currentThread(); //获取同步状态 state int c = getState(); //写锁状态 int w = exclusiveCount(c); //如果同步状态不为 0，说明有线程获得了读锁或写锁 if (c != 0) { //如果同步状态不为 0 ，并且写锁状态为 0，说明了读锁被占用，因读写锁互斥，故返回 false //若写锁状态不为 0，并且不是当前线程获得了写锁，则不能重入，返回 false if (w == 0 || current != getExclusiveOwnerThread()) return false; //如果超过了最大写锁数量，则抛出异常 if (w + exclusiveCount(acquires) > MAX_COUNT) throw new Error("Maximum lock count exceeded"); //若走到这一步，说明当前线程重入了，则计算重入次数，返回true setState(c + acquires); return true; } //到这说明 c 为 0，读锁和写锁都没有被占用 //如果写锁应该被阻塞或者 CAS 获取写锁失败，则返回false if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) return false; //把当前线程设为独占锁的所有者 setExclusiveOwnerThread(current); return true; } 写锁的释放
同理，写锁的释放从 unlock 方法开始，
public void unlock() { sync.release(1); } public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; } protected final boolean tryRelease(int releases) { //若独占锁的持有者不是当前线程，则抛出异常 if (!isHeldExclusively()) throw new IllegalMonitorStateException(); //每次释放，state 减 1 int nextc = getState() - releases; boolean free = exclusiveCount(nextc) == 0; if (free) setExclusiveOwnerThread(null); setState(nextc); return free; }可以看到，写锁的获取和释放和 ReentrantLock 的基本思想是差不多的。下面，着重讲解读锁的获取和释放，相对比较复杂。
读锁的获取
tryAcquireShared
从 ReadLock.lock 方法开始，
public void lock() { //调用 AQS 的方法 sync.acquireShared(1); } public final void acquireShared(int arg) { //如果 tryAcquireShared 方法返回小于 0，说明获取读锁失败 if (tryAcquireShared(arg) < 0) //以共享模式加入同步队列，再自旋抢锁 doAcquireShared(arg); } protected final int tryAcquireShared(int unused) { Thread current = Thread.currentThread(); int c = getState(); //如果有线程获取到了写锁，并且不是当前线程，则返回 -1 。 //这是因为，如果线程先获得了写锁，是可以重入再次获取读锁的，此为锁降级。 //否则不可重入。 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; //读锁数量 int r = sharedCount(c); //如果同时满足以下三个条件（读线程不应该被阻塞，读锁数量小于最大数量限制，CAS成功）， //则说明获取读锁成功，返回 1。然后再设置相关属性的值。 if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { //如果读锁状态为 0，说明还没有其他线程获取到读锁 if (r == 0) { //就把当前线程设置为第一个获取到读锁的线程 firstReader = current; //第一个读线程计数设置为 1 firstReaderHoldCount = 1; } else if (firstReader == current) { //如果当前线程是第一个获取读锁的线程，则重入，计数加 1 firstReaderHoldCount++; } else { //读锁状态不为 0，并且当前线程不是 firstReader //最近一个成功获取到读锁的线程计数器 HoldCounter rh = cachedHoldCounter; //如果计数器为空，或者计数器的 tid不是当前线程 id，说明有两种情况 //1.rh 还未被任何线程设置，此时只有 firstReader 一个线程获取到了读锁。 //2.rh 已经被设置了，并且不是当前线程，说明在当前线程之前，除了 firstReader， //还有其他线程获取到了读锁，那么当前线程就是第三个获取到读锁的（至少）。 if (rh == null || rh.tid != getThreadId(current)) //不管哪种情况，都需要创建并初始化当前线程的计数器，并赋值给 cachedHoldCounter //因为，当前线程是此时最后一个获取到读锁的线程，需要缓存下来 cachedHoldCounter = rh = readHolds.get(); //如果当前线程是最近一个获取到读锁的线程，并且计数为0， else if (rh.count == 0) //就把 rh 线程持有锁的次数信息，放入到本地线程 readHolds readHolds.set(rh); //最后把计数加 1 rh.count++; } return 1; } //若以上三个条件任意一个不满足，则调用此方法，再次全力尝试获取锁 return fullTryAcquireShared(current); }fullTryAcquireShared 这个方法和 tryAcquireShared 方法非常相似，只是多了一个自旋的过程，直到返回一个确定值（-1或1），才结束。
final int fullTryAcquireShared(Thread current) { HoldCounter rh = null; //自旋，直到返回一个确定值（1或 -1） for (;;) { int c = getState(); //如果写锁状态不为0，说明有线程获取到了写锁 if (exclusiveCount(c) != 0) { //获取到写锁的线程不是当前线程，则返回 -1 if (getExclusiveOwnerThread() != current) return -1; //这里省略了else，到这里说明了当前线程获取到了写锁，因此需要做锁降级处理， //把写锁降级为读锁。因为如果不这样做的话，线程就会阻塞到这，会导致死锁。 //然后跳转到 ①处继续执行 //===========// } else if (readerShouldBlock()) { //写锁空闲，并且读锁应该阻塞，说明 head.next正在等待获取写锁 //尽管读锁应该阻塞，但是此处也不应该立即阻塞，因为有可能存在读锁重入，需要再确认一下。 if (firstReader == current) {//当前线程是第一个读锁，可重入 // 将跳转到 ①处 } else { if (rh == null) { //第一次循环进来时肯定为 null rh = cachedHoldCounter; //取到缓存中最后一次获取到读锁的计数器 if (rh == null || rh.tid != getThreadId(current)) { rh = readHolds.get(); //计数为 0，说明当前线程没有获取到过读锁 if (rh.count == 0) //为了性能考虑，如果计数为 0，需要把它移除掉 readHolds.remove(); } } //走到这，说明当前线程不是 firstReader，也没有获取到过读锁，不符合重入条件， //那么就确定需要阻塞，只能去排队了，返回 -1 。 if (rh.count == 0) return -1; } } // ①处 //如果读锁数量达到了 MAX_COUNT，则抛出异常 if (sharedCount(c) == MAX_COUNT) throw new Error("Maximum lock count exceeded"); //CAS获取读锁，和 tryAcquireShared 的处理逻辑一样，不再赘述 if (compareAndSetState(c, c + SHARED_UNIT)) { if (sharedCount(c) == 0) { firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { if (rh == null) rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); else if (rh.count == 0) readHolds.set(rh); rh.count++; cachedHoldCounter = rh; // cache for release } return 1; } } }doAcquireShared
如果 tryAcquireShared 最终还是失败了，那么就执行 doAcquireShared 方法。
private void doAcquireShared(int arg) { //以共享模式加入同步队列 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == 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); } }setHeadAndPropagate
private void setHeadAndPropagate(Node node, int propagate) { //旧的头结点 Node h = head; //把当前 node 设置为新的头结点 setHead(node); //propagate 是 tryAcquireShared 方法的返回值 //若大于0，或者旧的头结点为空，或者头结点的 ws 小于0 //又或者新的头结点为空，或者新头结点的 ws 小于0，则获取后继节点 if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Node s = node.next; //没有后继节点或者后继节点是共享节点，就执行唤醒 if (s == null || s.isShared()) //释放掉资源，并唤醒后继节点，稍后讲解 doReleaseShared(); } }读锁的释放
tryReleaseShared
从 ReadLock.unlock方法开始，
public void unlock() { sync.releaseShared(1); } public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } protected final boolean tryReleaseShared(int unused) { Thread current = Thread.currentThread(); //当前线程为第一个读线程 if (firstReader == current) { //若 firstReader 的计数为1，则把它置为 null if (firstReaderHoldCount == 1) firstReader = null; else //否则，计数减 1，说明重入次数减 1 firstReaderHoldCount--; } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); int count = rh.count; if (count <= 1) { //如果当前线程的计数小于等于 1，则移除 readHolds.remove(); if (count <= 0) //若计数小于等于 0，则抛出异常 throw unmatchedUnlockException(); } //计数减 1 --rh.count; } for (;;) { int c = getState(); //读锁状态减 1，其实就是state值减 65536 //因为高16位的读锁实际值，在state中的表现就是相差 65536 int nextc = c - SHARED_UNIT; // CAS 设置 state 最新状态 if (compareAndSetState(c, nextc)) //如果读锁状态减为 0，就返回true //释放读锁对其它读线程没有任何影响， //但是如果读、写锁都空闲，就可以允许等待的写线程继续执行 return nextc == 0; } }doReleaseShared
如果 tryReleaseShared 方法返回 true，说明读锁释放成功，需要唤醒后继节点，
private void doReleaseShared() { for (;;) { //头结点 Node h = head; //说明队列中至少有两个节点 if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { //如果头结点的 ws 为 -1 ，则 CAS 把它设置为 0，因为唤醒后继节点后， //它就不需要做什么了。失败继续自旋尝试 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases // CAS 成功，则唤醒后继节点 unparkSuccessor(h); } //如果 ws 为 0，则把它设置为 -3 ，表明共享状态可向后传播，失败则继续自旋尝试 //后来我一直在想，为什么需要设置一个 PROPAGATE 这样的状态呢,但是还没头绪 //可以看下这篇文章分析，或许有一定的参考价值: //https://www.cnblogs.com/micrari/p/6937995.html //只能说 Doug Lea 大神的逻辑真是太缜密了，等我以后想明白了，再补充吧。 //可以暂时先理解为，这就是一个无条件传播的标志 else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } //如果此刻 h 等于头结点，说明头结点未改变，则跳出整个循环 //否则，说明头结点被其他线程修改过了，则继续下一次的循环判断 if (h == head) // loop if head changed break; } }结语
关于独占锁，比较简单。而读锁，涉及到了很多临界点和瞬时状态。其实细想，并不像表面上看起来那么简单，理解的会比较浅显，毕竟 Doug Lea 大神的思想不是常人能揣摩透的。
本篇只是我的一些个人理解，如有讲解不到位的地方，欢迎拍砖。
其实，还有很多细节问题，本文并没有展开。例如， setHeadAndPropagate 方法为什么判断两次新旧节点的 ws 状态，意义何为。 doReleaseShared 方法最后为什么需要设计 h == head 这样的判断，有什么含义。包括为什么要设计 PROPAGATE 状态，没有这个状态又如何。
看来路阻且长啊。。。以后再来补坑吧，这篇只能叫浅析了。￣□￣｜｜
如果本文对你有用，欢迎点赞，评论，转发。
学习是枯燥的，也是有趣的。我是「烟雨星空」，欢迎关注，可第一时间接收文章推送。
