【Java并发】揭秘Lock体系 -- 深入理解ReentrantReadWriteLock

系列文章目录

---

一、读写锁的介绍

如何在并发场景中解决线程安全的问题呢？在实际的业务开发中，几乎都会高频率地使用独占式锁解决并发场景问题，比如使用 Java 提供的 synchronized 关键字或 concurrent 包中实现了 Lock 接口的 ReentrantLock。它们都是独占式获取锁，也就是在同一时刻只有一个线程能够获取锁。而在一些业务场景中，大部分只是读数据，写数据的场景很少。如果仅仅是读数据，并不会影响数据的正确性，而如果在这种业务场景下依然使用独占式锁，很显然这将是出现性能瓶颈的地方，由于并发性能不好，会严重降低系统的吞吐量。针对这种读多写少的情况，Java 还提供了另外一个实现 Lock 接口的 ReentrantReadWriteLock（读写锁）。在分析 WriteLock 和 ReadLock 的互斥性时，可以分别按照 WriteLock 与 WriteLock 之间、WriteLock 与 ReadLock 之间和 ReadLock 与 ReadLock 之间三种情况进行分析。从读写锁的互斥性来看，读写锁允许同一时刻被多个读线程访问，但是在写线程访问时，所有的读线程和其他写线程都会被阻塞。

读写锁有哪些特性呢？

（1）公平性选择：支持非公平性（默认）和公平的锁获取方式，吞吐量还是非公平优于公平。
（2）重入性：支持重入，读锁获取后还能再次获取，但是当所有写锁未释放前，读锁是不能再重入获取到的。写锁获取之后能够再次获取写锁，同时也能够获取读锁。
（3）锁降级：遵循获取写锁、获取读锁、再释放写锁的次序，写锁能够降级为读锁。

要想彻底理解读写锁，必须能够理解以下几个问题

（1）读写锁是怎样实现分别记录读锁和写锁状态的？
（2）写锁是怎样获取和释放的？
（3）读锁是怎样获取和释放的？

带着这三个疑问，下面我们通过阅读源码的方式了解设计结构。

二、写锁详解

1. 写锁的获取

同步组件的实现聚合了同步器（AQS），并通过重写同步器中的方法实现同步组件的同步语义。写锁的实现也采用了这种方式。在同一时刻写锁不能被多个线程所获取，很显然写锁是独占式锁，而实现写锁的同步语义是通过重写 AQS 中的 tryAcquire () 方法实现的，源码如下：

protected final boolean tryAcquire(int acquires) {Thread current = Thread.currentThread();// 1. 获取写锁当前的同步状态int c = getState();// 2. 获取写锁获取的次数int w = exclusiveCount(c);if (c != 0) {// (Note: if c != 0 and w == 0 then shared count != 0)// 3.1 当读锁已被读线程获取，或者当前线程不是已获取写锁的线程// 当前线程获取写锁失败if (w == 0 || current != getExclusiveOwnerThread())return false;if (w + exclusiveCount(acquires) > MAX_COUNT)throw new Error("Maximum lock count exceeded");// Reentrant acquire// 3.2 当前线程获取写锁，支持可重复加锁setState(c + acquires);return true;}// 3.3 写锁未被任何线程获取，当前线程可获取写锁if (writerShouldBlock() ||!compareAndSetState(c, c + acquires))return false;setExclusiveOwnerThread(current);return true;
}

源码中有一个地方我们需要重点关注：exclusiveCount© 方法，该方法的源码如下：

static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; 
}

其中，EXCLUSIVE_MASK 被定义为 static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1。也就是说，EXCLUSIVE_MASK 为 1 左移 16 位然后减 1，即为 0x0000FFFF。而 exclusiveCount() 方法是将同步状态（state 为 int 类型）与 0x0000FFFF 相与，即取同步状态的低 16 位。

那么低 16 位代表什么呢？根据 exclusiveCount() 方法的注释，它代表独占式锁被获取的次数，即写锁被获取的次数。现在我们可以得出一个结论：同步状态的低 16 位是用来表示写锁的获取次数的。同时，还有一个方法值得我们注意：

static int sharedCount(int c){ return c >>> SHARED_SHIFT; 
}

该方法可以获取读锁被获取的次数，它主要是将同步状态（int c）右移 16 次，即取同步状态的高 16 位，现在我们又可以得出另外一个结论：同步状态的高 16 位是用来表示读锁被获取的次数的。

那么，至此我们就解答了，文章开篇提出的第一个问题：读写锁是怎样实现分别记录读锁和写锁状态的？

因此，写锁的 tryAcquire() 方法的主要逻辑为：当读锁已经被读线程获取或写锁已经被其他写线程获取，则写锁获取失败；否则，写锁获取成功并支持重入，增加写状态。

2. 写锁的释放

写锁释放的实现主要通过 AQS 的 tryRelease() 方法，源码如下：

protected final boolean tryRelease(int releases) {if (!isHeldExclusively())throw new IllegalMonitorStateException();//1. 同步状态减去写状态int nextc = getState() - releases;//2. 当前写状态是否为 0，为 0 则释放写锁boolean free = exclusiveCount(nextc) == 0;if (free)setExclusiveOwnerThread(null);//3. 不为 0 则更新同步状态setState(nextc);return free;
}

源码的具体实现可以参考注释。与 ReentrantLock 基本一致。这里需要注意的是要减少写状态（int nextc = getState ()-releases;），只需要用当前同步状态直接减去写状态即可，原因正是我们刚才所说的写状态是由同步状态的低 16 位表示的。

三、读锁详解

1.读锁的获取

在理解了写锁的基础上，再来深入理解读锁的底层原理是一件比较容易的事情。读锁是共享锁，即同一时刻该锁可以被多个读线程获取，获取锁的过程彼此不是互斥的。按照之前对 AQS 的介绍，共享式同步组件的同步语义需要通过重写 AQS 的 tryAcquireShared () 方法和 tryReleaseShared () 方法实现。读锁的获取方法如下：

protected final int tryAcquireShared(int unused) {Thread current = Thread.currentThread();int c = getState();//1. 如果写锁已经被获取并且获取写锁的线程不是当前线程，当前线程获取读锁失败，返回-1if (exclusiveCount(c) != 0 &&getExclusiveOwnerThread() != current)return -1;int r = sharedCount(c);if (!readerShouldBlock() &&r < MAX_COUNT &&//2. 当前线程获取读锁compareAndSetState(c, c + SHARED_UNIT)) {//3. 下面的代码可实现一些新增的功能，如getReadHoldCount()方法可用来获取读锁的次数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;}//4. 处理第2步操作失败的情况return fullTryAcquireShared(current);}

代码的主要逻辑可以参照注释。需要注意的是，当写锁被其他线程获取后，读锁获取失败；否则，获取成功后还会利用 CAS 更新同步状态。另外，由于同步状态的高 16 位表示读锁被获取的次数，因此在获取到读锁时会加上 SHARED_UNIT（(1 << SHARED_SHIFT) 即 0x00010000）。如果 CAS 失败或已获取读锁的线程再次获取读锁时，可以通过 fullTryAcquireShared () 方法实现。

2.读锁的释放

读锁的释放可以通过 tryReleaseShared () 方法实现，源码如下：

protected final boolean tryReleaseShared(int unused) {Thread current = Thread.currentThread();// 前面是为了实现getReadHoldCount等新功能if (firstReader == current) {// assert firstReaderHoldCount > 0;if (firstReaderHoldCount == 1)firstReader = null;elsefirstReaderHoldCount--;} else {HoldCounter rh = cachedHoldCounter;if (rh == null || rh.tid != getThreadId(current))rh = readHolds.get();int count = rh.count;if (count <= 1) {readHolds.remove();if (count <= 0)throw unmatchedUnlockException();}//4. 处理第2步操作失败的情况--rh.count;}for (; ; ) {int c = getState();// 读锁释放，将同步状态减去读状态即可int nextc = c - SHARED_UNIT;if (compareAndSetState(c, nextc))return nextc == 0;}}

代码逻辑可以参考注释，读锁和写锁的释放流程基本一致，在维护的同步状态中减去读标志位即可。

三、锁降级

读写锁支持锁降级，并且遵循获取写锁、获取读锁、再释放写锁的次序，写锁能够降级为读锁。但读写锁不支持锁升级。在ReentrantReadWriteLock源码中对锁降级给出一个示例，具体代码如下：

class CachedData {Object data;volatile boolean cacheValid;final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();void processCachedData() {rwl.readLock().lock();if (!cacheValid) {rwl.readLock().unlock();rwl.writeLock().lock();try {if (!cacheValid) {data = ...cacheValid = true;}rwl.readLock().lock();} finally {rwl.writeLock().unlock(); }}try {// 如果不进行锁降级操作，则可能存在数据并发的安全问题use(data);} finally {rwl.readLock().unlock();}}
}

cacheValid 是一个 volatile 变量，在数据层面上对多个线程都有可见性的要求，也就是说，对数据的处理变更都能被多个线程感知到。线程先获取到读锁后读取到状态标志量 cacheValid 为 false，随后会进行数据初始化操作。线程释放了读锁后，又获取写锁完成数据操作并且更新了 cacheValid 状态，由于当前线程持有写锁，所以此时其他线程会阻塞在读锁和写锁的 lock() 方法上等待获取到锁资源。当前线程完成数据操作后会继续获取读锁、随后释放写锁，这一整个过程就是锁降级的过程。那么锁降级的意义是什么呢？

在第 2 步获取到读锁后，由于读写锁的互斥性可以保证其他线程无法获取到写锁继续更新数据，从而保障了在当前线程的锁程里调用 use(data) 时数据是最新的，满足了数据的一致性和准确性。可以参考如下例子:

public class RwlDegrade {private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();private static ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();private static ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();private static int data = 0;private static void processData() {writeLock.lock();try {// 获取到写锁后进行数据处理System.out.println(Thread.currentThread().getName() + " get data:" + data);data++;} finally {writeLock.unlock();}try {// 模拟使用数据TimeUnit.SECONDS.sleep(1);System.out.println(Thread.currentThread().getName() + " process data:" + data);} catch (InterruptedException e) {e.printStackTrace();}}private static void degradeProcessData() {writeLock.lock();try {// 获取到写锁后进行数据处理System.out.println(Thread.currentThread().getName() + " get data:" + data);data++;// 关键的一步readLock.lock();} finally {writeLock.unlock();}try {// 使用数据TimeUnit.SECONDS.sleep(1);System.out.println(Thread.currentThread().getName() + " process data:" + data);} catch (InterruptedException e) {e.printStackTrace();} finally {readLock.unlock();}}public static void main(String[] args) {for (int i = 0; i < 10; i++) {Thread thread = new Thread(RwlDegrade::processData);thread.start();}}}

// 使用 processData () 方法时输出为：

Thread-0 get data:0
Thread-0 process data:10…
// 使用 degradeProcessData () 方法时输出为

Thread-0 get data:0
Thread-0 process data:1

通过上述示例代码可以清楚地看出，若当前线程不获取读锁，则数据可能会被其他线程获取到读锁后进行修改，导致当前线程使用的不是最新的数据。而严格按照锁降级的方式，再添加获取到读锁的操作后，就能够将数据维持到最新的数据视图，保障当前线程处理范围内不会出现数据并发安全的问题。

总结

本文主要通过源码的方式带大家学习ReentrantReadWriteLock，通过了解其实现的根本方法的源码来理解其原理。
以上就是本文全部内容，感谢各位能够看到最后，如有问题，欢迎各位大佬在评论区指正，希望大家可以有所收获！创作不易，希望大家多多支持！