@[toc]
概述
在上一篇博文jdk11源码-ReentrantLock源码中介绍了ReentrantLock的源码实现。ReentrantReadWriteLock是ReentrantLock的兄弟类，顾名思义，读写锁。当前其内部实现也是借助于AQS队列，不过与ReentrantLock的实现稍有不同，后面会逐步分析。
一个典型使用案例：
class RWDictionary {
private final Map<String, Data> m = new TreeMap<>();
private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
private final Lock r = rwl.readLock();
private final Lock w = rwl.writeLock();
public Data get(String key) {
r.lock();
try { return m.get(key); }
finally { r.unlock(); }
}
public List<String> allKeys() {
r.lock();
try { return new ArrayList<>(m.keySet()); }
finally { r.unlock(); }
}
public Data put(String key, Data value) {
w.lock();
try { return m.put(key, value); }
finally { w.unlock(); }
}
public void clear() {
w.lock();
try { m.clear(); }
finally { w.unlock(); }
}
}}读写标记的存储
在之前分析ReentrantLock时，讲述了线程获得锁的标记是在state上的，state=0表示没有被加锁，state=1表示加锁成功，state>1表示锁重入。
但是ReentrantReadWriteLock是读写锁，既要保存是否加锁，还要保存锁的类型，以及重入，只有state一个字段是无法满足的。所以ReentrantReadWriteLock对其进行了改造，一方面引入了HoldCounter来保存重入数量；一方面将state分为高低位，state是int类型，32位， 高16位保存共享锁（读锁），低16位保存独占锁（写锁）。接下来看一下详细实现。
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 6317671515068378041L;
//state被拆分为高低位，高16位保存共享锁（读锁），低16位保存独占锁（写锁）
//这几个字段是为了区分state的高低位的，具体使用会在下面讲解
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;
//返回共享锁的数量。c无符号右移16位也就是高16位的值。
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
// 返回独占锁的数量。EXCLUSIVE_MASK转为2进制是16个1，c是32位，进行与计算，高16位结果全为0，低16位结果与c的低16位相同，也就是c的低16位的值。
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
//HoldCounter类主要用于读锁的可重入，记录了重入的次数:count。
//HoldCounter会被包装在ThreadLocal中，是线程安全的，并且缓存在cachedHoldCounter变量中
static final class HoldCounter {
int count; // 初始值为0
// 这里使用id，而不是引用，目的是为了避免垃圾回收
final long tid = LockSupport.getThreadId(Thread.currentThread());
}
//这里重写了initialValue方法，可以保证获取到的HoldCounter对象是同一个，不会重复创建。
static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
/**
* 当前线程持有的 可重入读锁 的数量。
* 仅在构造函数和readObject方法中初始化。
* 当count的值降至0时删除。
*/
private transient ThreadLocalHoldCounter readHolds;
//缓存最后一个成功获取读锁的线程的HoldCounter
//通常来说，下一个要释放锁的线程就是最后一个获取锁的线程。这里使用cachedHoldCounter来单独保存，可以减少在ThreadLocal中查找的开销
private transient HoldCounter cachedHoldCounter;
//第一个获取读锁的线程，确切来说firstReader是最后一次将共享计数从0更改为1的唯一线程，并且从那时起就没有释放读锁定; 如果没有这样的线程，则返回null。
private transient Thread firstReader;
//第一个获取读锁线程的 hold count
private transient int firstReaderHoldCount;
Sync() {
//初始化一个readHolds，其count默认是0
readHolds = new ThreadLocalHoldCounter();
setState(getState()); // 这一步目的是为了确保state的可见性
}
//省略其他代码
}上面介绍了关键的一些属性变量及业务逻辑。关键点有
- state分为高16位和低16位，高16位保存共享锁（读锁），低16位保存独占锁（写锁）
- 使用HoldCounter 来记录读锁的重入次数，HoldCounter会被包装在ThreadLocal中，是线程安全的
另外，锁的获取与释放都是走的同一段代码逻辑。区别点在于AQS队列非空时的插入规则。
写锁的加解锁过程分析
writeLock.lock
public void lock() {
sync.acquire(1);
}调用AQS类的java.util.concurrent.locks.AbstractQueuedSynchronizer#acquire方法：
public final void acquire(int arg) {
if(!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}其中addWaiter和acquireQueued方法与ReentrantLock中分析的一样，这里就不在讲述。着重分析一下tryAcquire方。这里具体实现在java.util.concurrent.locks.ReentrantReadWriteLock.Sync#tryAcquire中：
@ReservedStackAccess
protectedfinal boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();//获取state的值
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;//走到这里，c不等于0，w（低16位）是0，所以高16位肯定有值。也就是说此时有共享锁存在。
if (w + exclusiveCount(acquires) > MAX_COUNT)//校验独占锁的数量是否超标
throw new Error("Maximum lock count exceeded");
//走到这里，说明当前是写锁重入。因为走到这一步只可能是：c不是0，w不是0，exclusiveOwnerThread标识的线程是当前线程。
setState(c + acquires);
return true;
}
//走到这里说明c的值是0。也就是读锁和写锁都没有。当然可以加锁了。
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);//加锁成功，设置获取独占锁的线程为当前线程
return true;
}上面就是获取锁的过程，总结一下就是：
- state的值是0，说明当前没有锁，那么判断写锁是否应该阻塞，如果不阻塞，则修改state的值，修改成功则加锁成功
- state的值不是0，低16位是0，说明高16位不是0，此时有共享锁存在，不可以加锁。
- state的值不是0，低16位不是0，但是当前获取独占锁的线程不是当前线程，不可以加锁
- state的值不是0，低16位不是0，当前获取独占锁的线程是当前线程，说明是写锁重入，可以加锁
writeLock.unlock
writeLock.unlock源码：
public void unlock() {
sync.release(1);
}他调用AQS类的release方法：
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}分析tryRelease方法，他的实现在ReentrantReadWriteLock.Sync类中， 其他的与ReentranteLock中分析的一致
protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())//独占锁的持有者不是当前线程，抛异常
throw new IllegalMonitorStateException();
int nextc = getState() - releases;//这里是独占锁，低16位，所以可以直接相减，下一步会校验是不是低16位减为0
boolean free = exclusiveCount(nextc) == 0;//判断低16位的独占锁是否都释放完
if (free)
setExclusiveOwnerThread(null);//独占锁已经释放完，exclusiveOwnerThread设为null
setState(nextc);//释放state
return free;
}独占锁的释放过程比较简单的，首先检查加锁的是不是当前线程，不是则抛异常。接下来校验释放releases数量的state后，低16位是否为0，是，则说明独占锁释放完成，固设置exclusiveOwnerThread=null，否则设置新的state。
接下来看一下unparkSuccessor方法，写锁释放后，唤醒队列中后继节点，该后继节点可能是读锁也可能是写锁。
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
node.compareAndSetWaitStatus(ws, 0);//如果<0（这里一般是Node.PROPAGATE这种状态），重新将其waitstatus设置为0
Node s = node.next;
if (s == null || s.waitStatus > 0) {//如果后继节点被取消，那么找到后继第一个waitStatus <= 0的节点，将其唤醒
s = null;
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}公平锁与非公平锁
在上面tryAcquire代码中有一个方法writerShouldBlock来判断写锁是否应该被阻塞。这里公平锁和非公平锁的实现是不同的。
公平锁的实现：
static final class FairSync extends Sync {
private static final long serialVersionUID = -2274990926593161451L;
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
}
public final boolean hasQueuedPredecessors() {
Node h, s;
if ((h = head) != null) {
if ((s = h.next) == null || s.waitStatus > 0) {
s = null; // traverse in case of concurrent cancellation
for (Node p = tail; p != h && p != null; p = p.prev) {
if (p.waitStatus <= 0)
s = p;
}
}
if (s != null && s.thread != Thread.currentThread())
return true;
}
return false;
}读锁的加解锁过程分析
readLock.lock
public void lock() {
sync.acquireShared(1);
}该方法会调用AQS的acquireShared方法
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}首先尝试获取tryAcquireShared，如果失败则执行doAcquireShared。
tryAcquireShared
tryAcquireShared的实现在ReentrantReadWriteLock中：
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;//已经有写锁获取，读锁加锁失败
int r = sharedCount(c);//共享锁的数量
if (!readerShouldBlock() && r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {//不阻塞，并且申请一个读锁成功
if (r == 0) {//共享锁数量为0：则记录最后一次将共享锁计数从0更改为1的线程，初始化firstReaderHoldCount =1
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;//firstReader 标识的第一个读线程 读锁重入
} else {
HoldCounter rh = cachedHoldCounter;//cachedHoldCounter：缓存的最后一个获取读锁的线程的HoldCounter
if (rh == null || rh.tid != LockSupport.getThreadId(current))//缓存为空，或者缓存的线程id不是当前线程
cachedHoldCounter = rh = readHolds.get();//获取threadlocal中当前线程的计数器缓存HoldCounter ，并将其添加到缓存cachedHoldCounter中
else if (rh.count == 0)//由于count为0时，readHolds中的缓存会被移除掉，所以这里要在set一下
readHolds.set(rh);//将缓存的cachedHoldCountert添加到threadlocal变量readHolds中
rh.count++;//当前线程重入计数器加一
}
return 1;
}
return fullTryAcquireShared(current);
}可以看到这里入参没有用到，而是通过compareAndSetState(c, c + SHARED_UNIT)固定的申请一个读锁标记。整体流程如下：
==这里隐藏的优化点==：
- firstReader 和firstReaderHoldCount。用于缓存第一个获取读锁的线程。目的是为了减少threadlocal的操作，提高效率。
- cachedHoldCounter：用于缓存最后一个获取读锁的线程。当然这个也会同步保存到threadlocal变量readHolds中
- 将大概率执行成功的代码与通用代码分隔，提高性能（可能会降低一点可读性）。
当(!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)判断失败时，会进入fullTryAcquireShared()方法，该方法会采用CAS的方式来保证操作执行成功。
fullTryAcquireShared方法与上面tryAcquireShared逻辑基本相同，但是为什么也要写两份呢？其实这也是为了性能考虑。
读者在阅读java线程及锁相关的代码时，会经常遇到类似的逻辑，将通用方法与大概率出现的代码逻辑分开，对于大概率出现的逻辑进行针对性的优化来提高性能。
读者可以考虑将fullTryAcquireShared上面的代码去除，代码运行并不会受影响。
final int fullTryAcquireShared(Thread current) {
HoldCounter rh = null;
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0) {
if (getExclusiveOwnerThread() != current)//有其他线程加了独占锁，加锁失败
return -1;
} else if (readerShouldBlock()) {
//读锁需要被阻塞
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
} else {
if (rh == null) {
rh = cachedHoldCounter;
if (rh == null ||
rh.tid != LockSupport.getThreadId(current)) {
rh = readHolds.get();
if (rh.count == 0)
readHolds.remove();//count为0，当前线程的所有读锁都释放了，将其删除
}
}
if (rh.count == 0)
return -1;
}
}
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
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 != LockSupport.getThreadId(current))
rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
cachedHoldCounter = rh; //缓存最后一个成功获取读锁的线程
}
return 1;
}
}
}doAcquireShared
先看一下tryAcquireShared方法有哪些情况会返回<0
- 有独占锁存在，并且该独占锁不是当前线程加的。
- 没有独占锁存在，并且读锁需要等待时，并且当前线程不是第一个获取读锁的线程，并且最后一个获取锁的线程全部已经释放锁时。
执行到doAcquireShared这个方法，该线程加入AQS队列了，队列中的读锁成功获取锁后，需要唤醒后继节点，具体后面会详细介绍。
doAcquireShared: 以共享无中断模式获取。无限循环获取读锁，直到成功或者发生异常
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);//以共享模式添加到AQS队列尾部。
boolean interrupted = false;
try {
for (;;) {
final Node p = node.predecessor();//获取上一个node
if (p == head) {//上一个节点是head节点，说明轮到自己获取锁了，则尝试获取读锁tryAcquireShared
//翻阅源码，在ReentrantReadWriteLock中，这个返回值 r 要么是1，要么是-1
//当然其他情况，可能返回0
int r = tryAcquireShared(arg);
if (r >= 0) {//获取读锁成功。
setHeadAndPropagate(node, r);
p.next = null; // 老的头结点已经没有用了，其next指向null，目的为了GC释放内存
return;
}
}
if (shouldParkAfterFailedAcquire(p, node))//检测是否应该park等待
interrupted |= parkAndCheckInterrupt();
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
} finally {
if (interrupted)
selfInterrupt();//被中断过，重置中断标记
}
}
//设置头结点并且广播 唤醒后继节点
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
setHead(node);
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())//下一个节点如果为空，或者是共享模式的节点
doReleaseShared();
}
}共享模式的Node在新建时，会设置nextWaiter=SHARE：
Node(Node nextWaiter) {
this.nextWaiter = nextWaiter;
THREAD.set(this, Thread.currentThread());
}这里需要说一下tryAcquireShared的返回值。 结合ReentrantReadWriteLock、CountDownLatch、Semaphore等实现，tryAcquireShared的返回值可能是0,1，-1，大于1 四种情况，其中大于1可以和1归为一类 。
==tryAcquireShared的返回值大于0时，表示后继节点需要被唤醒；0表示加锁成功但是不需要唤醒后继节点；-1表示失败==
再分析一下setHeadAndPropagate方法唤醒规则(propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0)：
- propagate>0表示后继节点需要被唤醒。
- 或者不论旧的头结点还是新的头结点，只要他的waitStatus<0那么就执行唤醒
这个waitStatus的设置参考shouldParkAfterFailedAcquire方法，其源码在上一篇文章jdk11源码-ReentrantLock源码已经分析过，核心就是在park时首先将其前驱节点的waitStatus设置为SIGNAL（-1）。所以waitStatus<0时就执行唤醒。
注意一点区别：如果tryAcquireShared成功，此时并没有加入AQS队列中，所以不会有通知唤醒之类的操作。而到了doAcquireShared方法中，首先是要现将其加入到AQS队列中，所以在获取所以后，就需要唤醒后继节点了。
doReleaseShared
private void doReleaseShared() {
for (;;) {
Node h = head;//注意这里已经是新的头节点了
if (h != null && h != tail) {//头结点存在并且不等于tail，也就是AQS队列不为空的情况
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {//是SIGNAL状态，则需要unpark后继节点
if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))//将头结点的waitstatue设置为0，以后就不会再次唤醒后继节点了。这一步是为了解决并发问题，保证只unpark一次！！
continue; // 设置失败则进行重试
unparkSuccessor(h);//唤醒头节点的后继节点
}
else if (ws == 0 && !h.compareAndSetWaitStatus(0, Node.PROPAGATE))//这里尝试将头结点设置为PROPAGATE。保证传播性
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
node.compareAndSetWaitStatus(ws, 0);//如果<0（这里一般是Node.PROPAGATE这种状态），重新将其waitstatus设置为0
Node s = node.next;
if (s == null || s.waitStatus > 0) {//如果后继节点被取消，那么找到后继第一个waitStatus <= 0的节点，将其唤醒
s = null;
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}读锁的特性之一：AQS队列中读锁获取成功后，需要将其后面的读锁唤醒。当后继读锁的节点唤醒后，他会重新申请读锁，此时如果成功，那么他会再次唤醒他的后继节点，如此往复，AQS队列中的头结点后的所有读锁都会被唤醒，如果遇到等待的写锁，终止传递。
总结读锁加锁过程：
- acquireShared()申请锁，如果成功，则进入临界区继续执行。如果失败，则加入AQS等待队列，挂起（park）等待被唤醒（unpark）。
- AQS队列中的读锁被唤醒后，重新尝试获取锁，如果成功，则进入临界区执行，并且唤醒后继共享节点，并且会传递下去一次唤醒所有的共享节点，直到队尾或者遇到写锁。
readLock.unlock
public void unlock() {
sync.releaseShared(1);
}
//releaseShared调用AQS类中的releaseShared方法
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {//尝试释放锁，ReentrantReadWriteLock中，只有读锁和写锁全部释放才会返回true
doReleaseShared();//如果释放锁成功后，没有读锁也没有写锁，那么唤醒AQS的后继节点
return true;
}
return false;
}releaseShared方法很简单，先释放锁，在唤醒后继节点。
ReentrantReadWriteLock中，tryReleaseShared方法只有读锁和写锁全部释放才会返回true。这里注意：tryReleaseShared是一个接口，不同的锁的实现机制不同，比如信号量（Semaphore）机制下，只要释放资源成功就会返回true，而不会关心state的余额。这个在分析Semaphore源码时具体分析。
我们看ReentrantReadWriteLock中tryReleaseShared的实现：
protected final boolean tryReleaseShared(int unused) {
Thread current = Thread.currentThread();
if (firstReader == current) {//当前线程是第一个获取读锁线程，释放 firstReader 及firstReaderHoldCount
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
} else {
//当前线程不是第一个读线程，需要从threadlocal中获取HoldCounter ，释放其count
HoldCounter rh = cachedHoldCounter;
if (rh == null ||
rh.tid != LockSupport.getThreadId(current))
rh = readHolds.get();
int count = rh.count;
if (count <= 1) {
readHolds.remove();//说明这已经是最后一次释放锁了，直接删除即可
if (count <= 0)
throw unmatchedUnlockException();//说明锁计数器出现异常
}
--rh.count;//释放count的值
}
for (;;) {
int c = getState();
int nextc = c - SHARED_UNIT;//读锁标记减去 1 （还记得高16位表示读锁数量吗？）
if (compareAndSetState(c, nextc))
//当没有读锁也没有写锁时，返回true
return nextc == 0;
}
}tryReleaseShared方法释放读锁占用的资源，释放以后，如果state!=0，说明有读锁或者写锁。那么直接返回。如果state==0，说明此事没有读锁也没有写锁了，需要唤醒AQS队列中等待的节点。
doReleaseShared有两个入口：一个是读锁释放资源后，一个是setHeadAndPropagate方法中。doReleaseShared的源码上面已经分析过。这里说一下不同之处：前面是读锁加锁后setHeadAndPropagate内部调用的doReleaseShared，他前面有一个判断s == null || s.isShared()，也就是说只能后继节点是读锁才可以唤醒后继节点。这也很好理解，一个读锁抢占资源成功，那么AQS队列中后面等待的读锁都应该唤醒进入临界区，这才符合读锁（共享锁）的特性吗。
然而在读锁释放资源时调用doReleaseShared方法时，此时唤醒的有可能是读锁，也可能是写锁，因为我们并不知道AQS队列中第一个节点是读还是写，所以根据唤醒的节点不同走不同的分支。
