读写锁 ReentrantReadWriteLock
首先我们来了解一下 ReentrantReadWriteLock 的作用是什么？和 ReentranLock 有什么区别？Reentrancy 英文的意思是可重入性。ReentrantReadWriteLock下文简称（rrwl）
下面总结来自 Java并发编程—ReentrantReadWriteLock ，你也可以从JDK 中阅读到这段。
ReentrantReadWriteLock是Lock的另一种实现方式，我们已经知道了ReentrantLock是一个排他锁，同一时间只允许一个线程访问，而ReentrantReadWriteLock允许多个读线程同时访问，但不允许写线程和读线程、写线程和写线程同时访问。相对于排他锁，提高了并发性。在实际应用中，大部分情况下对共享数据（如缓存）的访问都是读操作远多于写操作，这时ReentrantReadWriteLock能够提供比排他锁更好的并发性和吞吐量。
ReentrantReadWriteLock支持以下功能：
1）支持公平和非公平的获取锁的方式；
2）支持可重入。读线程在获取了读锁后还可以获取读锁；写线程在获取了写锁之后既可以再次获取写锁又可以获取读锁；
3）还允许从写入锁降级为读取锁，其实现方式是：先获取写入锁，然后获取读取锁，最后释放写入锁。但是，从读取锁升级到写入锁是不允许的；（这也说明了在rrwl中写锁的等级高于读锁）
4）读取锁和写入锁都支持锁获取期间的中断；
5）Condition支持。仅写入锁提供了一个 Conditon 实现；读取锁不支持 Conditon ，readLock().newCondition() 会抛出 UnsupportedOperationException。
下面是两个来自文档的Demo
1 class CachedData {
2 Object data;
3 volatile boolean cacheValid;
4 final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
5
6 void processCachedData() {
7 rwl.readLock().lock();
8 if (!cacheValid) {
9 // Must release read lock before acquiring write lock
10 rwl.readLock().unlock();
11 rwl.writeLock().lock();
12 try {
13 // Recheck state because another thread might have
14 // acquired write lock and changed state before we did.
15 if (!cacheValid) {
16 data = ...
17 cacheValid = true;
18 }
19 // Downgrade by acquiring read lock before releasing write lock
20 rwl.readLock().lock();
21 } finally {
22 rwl.writeLock().unlock(); // Unlock write, still hold read
23 }
24 }
25
26 try {
27 use(data);
28 } finally {
29 rwl.readLock().unlock();
30 }
31 }
32 }
在一些集合的使用上，rrwl 可以用来提升并发性。
1 class RWDictionary {
2 private final Map<String, Data> m = new TreeMap<String, Data>();
3 private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
4 private final Lock r = rwl.readLock();
5 private final Lock w = rwl.writeLock();
6
7 public Data get(String key) {
8 r.lock();
9 try { return m.get(key); }
10 finally { r.unlock(); }
11 }
12 public String[] allKeys() {
13 r.lock();
14 try { return m.keySet().toArray(); }
15 finally { r.unlock(); }
16 }
17 public Data put(String key, Data value) {
18 w.lock();
19 try { return m.put(key, value); }
20 finally { w.unlock(); }
21 }
22 public void clear() {
23 w.lock();
24 try { m.clear(); }
25 finally { w.unlock(); }
26 }
27 }
ReentrantReadWriteLock 概述
我们之前分析AQS 知道了，在AQS中使用的两种模式，一种是独占模式，一种是共享模式。
（图片来源见参考资料）
AQS 的精髓在于内部的属性 state：
- 对于独占模式来说，通常就是 0 代表可获取锁，1 代表锁被别人获取了，重入例外
- 而共享模式下，每个线程都可以对 state 进行加减操作
也就是说，独占模式和共享模式对于 state 的操作完全不一样，那读写锁 ReentrantReadWriteLock 中是怎么使用 state 的呢？答案是将 state 这个 32 位的 int 值分为高 16 位和低 16位，分别用于共享模式和独占模式。
ReentrantReadWriteLock 源码分析
下文部分源码分析来自参考资料，摘抄过来只是为了纪录。
1 abstract static class Sync extends AbstractQueuedSynchronizer {
2 // 下面这块说的就是将 state 一分为二，高 16 位用于共享模式，低16位用于独占模式
3 static final int SHARED_SHIFT = 16;
4 static final int SHARED_UNIT = (1 << SHARED_SHIFT);
5 static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
6 static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
7 // 取 c 的高 16 位值，代表读锁的获取次数(包括重入)
8 static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
9 // 取 c 的低 16 位值，代表写锁的重入次数，因为写锁是独占模式
10 static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
11
12 // 这个嵌套类的实例用来记录每个线程持有的读锁数量(读锁重入)
13 static final class HoldCounter {
14 // 持有的读锁数
15 int count = 0;
16 // 线程 id
17 final long tid = getThreadId(Thread.currentThread());
18 }
19
20 // ThreadLocal 的子类
21 static final class ThreadLocalHoldCounter
22 extends ThreadLocal<HoldCounter> {
23 public HoldCounter initialValue() {
24 return new HoldCounter();
25 }
26 }
27 /**
28 * 组合使用上面两个类，用一个 ThreadLocal 来记录当前线程持有的读锁数量
29 */
30 private transient ThreadLocalHoldCounter readHolds;
31
32 // 用于缓存，记录"最后一个获取读锁的线程"的读锁重入次数，
33 // 所以不管哪个线程获取到读锁后，就把这个值占为已用，这样就不用到 ThreadLocal 中查询 map 了
34 // 算不上理论的依据：通常读锁的获取很快就会伴随着释放，
35 // 显然，在 获取->释放 读锁这段时间，如果没有其他线程获取读锁的话，此缓存就能帮助提高性能
36 private transient HoldCounter cachedHoldCounter;
37
38 // 第一个获取读锁的线程(并且其未释放读锁)，以及它持有的读锁数量
39 private transient Thread firstReader = null;
40 private transient int firstReaderHoldCount;
41
42 Sync() {
43 // 初始化 readHolds 这个 ThreadLocal 属性
44 readHolds = new ThreadLocalHoldCounter();
45 // 为了保证 readHolds 的内存可见性
46 setState(getState()); // ensures visibility of readHolds
47 }
48 ...
49 }
注意一下，里面有个HoldCounter 是静态内部类，所以外部只要引用到它就会改变到 readHolds 的值。在读锁获取锁的时候会讲到。
- readHolds : 某个线程获取读锁的数量，用来记录某个线程的重入锁数量。他是数据结构是ThreadLocalHoldCounter ，而ThreadLocalHoldCounter 是继承 ThreadLocal 。
- cachedHoldCounter ： 可以从名字看出来是缓存的作用，它代表的是最后一个线程成功获取读锁的数量。它可以起到缓存的作用的依据是刚释放的锁又会很快地去获取锁。
1 // ReadLock
2 public void lock() {
3 sync.acquireShared(1);
4 }
5 // AQS
6 public final void acquireShared(int arg) {
7 if (tryAcquireShared(arg) < 0)
8 doAcquireShared(arg);
9 }
上面的代码中，要进入 if 分支，需要满足：readerShouldBlock() 返回 false，并且 CAS 要成功（我们先不要纠结 MAX_COUNT 溢出。
那我们反向推，怎么样进入到最后的 fullTryAcquireShared ：
- readerShouldBlock() 返回 true，2 种情况：
- 在 FairSync 中说的是 hasQueuedPredecessors()，即阻塞队列中有其他元素在等待锁。
也就是说，公平模式下，有人在排队呢，你新来的不能直接获取锁
- 在 NonFairSync 中说的是 apparentlyFirstQueuedIsExclusive()，即判断阻塞队列中 head 的第一个后继节点是否是来获取写锁的，如果是的话，让这个写锁先来，避免写锁饥饿。
作者给写锁定义了更高的优先级，所以如果碰上获取写锁的线程马上就要获取到锁了，获取读锁的线程不应该和它抢。
如果 head.next 不是来获取写锁的，那么可以随便抢，因为是非公平模式，大家比比 CAS 速度
- 在 FairSync 中说的是 hasQueuedPredecessors()，即阻塞队列中有其他元素在等待锁。
- compareAndSetState(c, c + SHARED_UNIT) 这里 CAS 失败，存在竞争。可能是和另一个读锁获取竞争，当然也可能是和另一个写锁获取操作竞争。
然后就会来到 fullTryAcquireShared 中再次尝试：
1 /**
2 * 1. 刚刚我们说了可能是因为 CAS 失败，如果就此返回，那么就要进入到阻塞队列了，
3 * 想想有点不甘心，因为都已经满足了 !readerShouldBlock()，也就是说本来可以不用到阻塞队列的，
4 * 所以进到这个方法其实是增加 CAS 成功的机会
5 * 2. 在 NonFairSync 情况下，虽然 head.next 是获取写锁的，我知道它等待很久了，我没想和它抢，
6 * 可是如果我是来重入读锁的，那么只能表示对不起了
7 */
8 final int fullTryAcquireShared(Thread current) {
9 HoldCounter rh = null;
10 // 别忘了这外层有个 for 循环
11 for (;;) {
12 int c = getState();
13 // 如果其他线程持有了写锁，自然这次是获取不到读锁了，乖乖到阻塞队列排队吧
14 if (exclusiveCount(c) != 0) {
15 if (getExclusiveOwnerThread() != current)
16 return -1;
17 // else we hold the exclusive lock; blocking here
18 // would cause deadlock.
19 } else if (readerShouldBlock()) {
20 /**
21 * 进来这里，说明：
22 * 1. exclusiveCount(c) == 0：写锁没有被占用
23 * 2. readerShouldBlock() 为 true，说明阻塞队列中有其他线程在等待
24 *
25 * 既然 should block，那进来这里是干什么的呢？
26 * 答案：是进来处理读锁重入的！
27 *
28 */
29
30 // firstReader 线程重入读锁，直接到下面的 CAS
31 if (firstReader == current) {
32 // assert firstReaderHoldCount > 0;
33 } else {
34 if (rh == null) {
35 rh = cachedHoldCounter;
36 if (rh == null || rh.tid != getThreadId(current)) {
37 // cachedHoldCounter 缓存的不是当前线程
38 // 那么到 ThreadLocal 中获取当前线程的 HoldCounter
39 // 如果当前线程从来没有初始化过 ThreadLocal 中的值，get() 会执行初始化
40 rh = readHolds.get();
41 // 如果发现 count == 0，也就是说，纯属上一行代码初始化的，那么执行 remove
42 // 然后往下两三行，乖乖排队去
43 if (rh.count == 0)
44 readHolds.remove();
45 }
46 }
47 if (rh.count == 0)
48 // 排队去。
49 return -1;
50 }
51 /**
52 * 这块代码我看了蛮久才把握好它是干嘛的，原来只需要知道，它是处理重入的就可以了。
53 * 就是为了确保读锁重入操作能成功，而不是被塞到阻塞队列中等待
54 *
55 * 另一个信息就是，这里对于 ThreadLocal 变量 readHolds 的处理：
56 * 如果 get() 后发现 count == 0，居然会做 remove() 操作，
57 * 这行代码对于理解其他代码是有帮助的
58 */
59 }
60
61 if (sharedCount(c) == MAX_COUNT)
62 throw new Error("Maximum lock count exceeded");
63
64 if (compareAndSetState(c, c + SHARED_UNIT)) {
65 // 这里 CAS 成功，那么就意味着成功获取读锁了
66 // 下面需要做的是设置 firstReader 或 cachedHoldCounter
67
68 if (sharedCount(c) == 0) {
69 // 如果发现 sharedCount(c) 等于 0，就将当前线程设置为 firstReader
70 firstReader = current;
71 firstReaderHoldCount = 1;
72 } else if (firstReader == current) {
73 firstReaderHoldCount++;
74 } else {
75 // 下面这几行，就是将 cachedHoldCounter 设置为当前线程
76 if (rh == null)
77 rh = cachedHoldCounter;
78 if (rh == null || rh.tid != getThreadId(current))
79 rh = readHolds.get();
80 else if (rh.count == 0)
81 readHolds.set(rh);
82 rh.count++;
83 cachedHoldCounter = rh;
84 }
85 // 返回大于 0 的数，代表获取到了读锁
86 return 1;
87 }
88 }
89 }
读锁释放
下面我们看看读锁释放的流程：
1 // ReadLock
2 public void unlock() {
3 sync.releaseShared(1);
4 }
5 // Sync
6 public final boolean releaseShared(int arg) {
7 if (tryReleaseShared(arg)) {
8 doReleaseShared(); // 这句代码其实唤醒 获取写锁的线程，往下看就知道了
9 return true;
10 }
11 return false;
12 }
13
14 // Sync
15 protected final boolean tryReleaseShared(int unused) {
16 Thread current = Thread.currentThread();
17 if (firstReader == current) {
18 if (firstReaderHoldCount == 1)
19 // 如果等于 1，那么这次解锁后就不再持有锁了，把 firstReader 置为 null，给后来的线程用
20 // 为什么不顺便设置 firstReaderHoldCount = 0？因为没必要，其他线程使用的时候自己会设值
21 firstReader = null;
22 else
23 firstReaderHoldCount--;
24 } else {
25 // 判断 cachedHoldCounter 是否缓存的是当前线程，不是的话要到 ThreadLocal 中取
26 HoldCounter rh = cachedHoldCounter;
27 if (rh == null || rh.tid != getThreadId(current))
28 rh = readHolds.get();
29
30 int count = rh.count;
31 if (count <= 1) {
32
33 // 这一步将 ThreadLocal remove 掉，防止内存泄漏。因为已经不再持有读锁了
34 readHolds.remove();
35
36 if (count <= 0)
37 // 就是那种，lock() 一次，unlock() 好几次的逗比
38 throw unmatchedUnlockException();
39 }
40 // count 减 1
41 --rh.count;
42 }
43
44 for (;;) {
45 int c = getState();
46 // nextc 是 state 高 16 位减 1 后的值
47 int nextc = c - SHARED_UNIT;
48 if (compareAndSetState(c, nextc))
49 // 如果 nextc == 0，那就是 state 全部 32 位都为 0，也就是读锁和写锁都空了
50 // 此时这里返回 true 的话，其实是帮助唤醒后继节点中的获取写锁的线程
51 return nextc == 0;
52 }
53 }
54
读锁释放的过程还是比较简单的，主要就是将 hold count 减 1，如果减到 0 的话，还要将 ThreadLocal 中的 remove 掉。然后是在 for 循环中将 state 的高 16 位减 1，如果发现读锁和写锁都释放光了，那么唤醒后继的获取写锁的线程。
写锁获取
- 写锁是独占锁。
- 如果有读锁被占用，写锁获取是要进入到阻塞队列中等待的。
1 // ReadLock
2 public void unlock() {
3 sync.releaseShared(1);
4 }
5 // Sync
6 public final boolean releaseShared(int arg) {
7 if (tryReleaseShared(arg)) {
8 doReleaseShared(); // 这句代码其实唤醒 获取写锁的线程，往下看就知道了
9 return true;
10 }
11 return false;
12 }
13
14 // Sync
15 protected final boolean tryReleaseShared(int unused) {
16 Thread current = Thread.currentThread();
17 if (firstReader == current) {
18 if (firstReaderHoldCount == 1)
19 // 如果等于 1，那么这次解锁后就不再持有锁了，把 firstReader 置为 null，给后来的线程用
20 // 为什么不顺便设置 firstReaderHoldCount = 0？因为没必要，其他线程使用的时候自己会设值
21 firstReader = null;
22 else
23 firstReaderHoldCount--;
24 } else {
25 // 判断 cachedHoldCounter 是否缓存的是当前线程，不是的话要到 ThreadLocal 中取
26 HoldCounter rh = cachedHoldCounter;
27 if (rh == null || rh.tid != getThreadId(current))
28 rh = readHolds.get();
29
30 int count = rh.count;
31 if (count <= 1) {
32
33 // 这一步将 ThreadLocal remove 掉，防止内存泄漏。因为已经不再持有读锁了
34 readHolds.remove();
35
36 if (count <= 0)
37 // 就是那种，lock() 一次，unlock() 好几次的逗比
38 throw unmatchedUnlockException();
39 }
40 // count 减 1
41 --rh.count;
42 }
43
44 for (;;) {
45 int c = getState();
46 // nextc 是 state 高 16 位减 1 后的值
47 int nextc = c - SHARED_UNIT;
48 if (compareAndSetState(c, nextc))
49 // 如果 nextc == 0，那就是 state 全部 32 位都为 0，也就是读锁和写锁都空了
50 // 此时这里返回 true 的话，其实是帮助唤醒后继节点中的获取写锁的线程
51 return nextc == 0;
52 }
53 }
54
下面看一眼 writerShouldBlock() 的判定，然后你再回去看一篇写锁获取过程。
1 static final class NonfairSync extends Sync {
2 // 如果是非公平模式，那么 lock 的时候就可以直接用 CAS 去抢锁，抢不到再排队
3 final boolean writerShouldBlock() {
4 return false; // writers can always barge
5 }
6 ...
7 }
8 static final class FairSync extends Sync {
9 final boolean writerShouldBlock() {
10 // 如果是公平模式，那么如果阻塞队列有线程等待的话，就乖乖去排队
11 return hasQueuedPredecessors();
12 }
13 ...
14 }
15
写锁释放
1 // WriteLock
2 public void unlock() {
3 sync.release(1);
4 }
5
6 // AQS
7 public final boolean release(int arg) {
8 // 1. 释放锁
9 if (tryRelease(arg)) {
10 // 2. 如果独占锁释放"完全"，唤醒后继节点
11 Node h = head;
12 if (h != null && h.waitStatus != 0)
13 unparkSuccessor(h);
14 return true;
15 }
16 return false;
17 }
18
19 // Sync
20 // 释放锁，是线程安全的，因为写锁是独占锁，具有排他性
21 // 实现很简单，state 减 1 就是了
22 protected final boolean tryRelease(int releases) {
23 if (!isHeldExclusively())
24 throw new IllegalMonitorStateException();
25 int nextc = getState() - releases;
26 boolean free = exclusiveCount(nextc) == 0;
27 if (free)
28 setExclusiveOwnerThread(null);
29 setState(nextc);
30 // 如果 exclusiveCount(nextc) == 0，也就是说包括重入的，所有的写锁都释放了，
31 // 那么返回 true，这样会进行唤醒后继节点的操作。
32 return free;
33 }
锁降级
将持有写锁的线程，去获取读锁的过程称为锁降级（Lock downgrading）。这样，此线程就既持有写锁又持有读锁。但是，锁升级是不可以的。线程持有读锁的话，在没释放的情况下不能去获取写锁，因为会发生死锁。回去看下写锁获取的源码：
1 protected final boolean tryAcquire(int acquires) {
2
3 Thread current = Thread.currentThread();
4 int c = getState();
5 int w = exclusiveCount(c);
6 if (c != 0) {
7 // 看下这里返回 false 的情况：
8 // c != 0 && w == 0: 写锁可用，但是有线程持有读锁(也可能是自己持有)
9 // c != 0 && w !=0 && current != getExclusiveOwnerThread(): 其他线程持有写锁
10 // 也就是说，只要有读锁或写锁被占用，这次就不能获取到写锁
11 if (w == 0 || current != getExclusiveOwnerThread())
12 return false;
13 ...
14 }
15 ...
16 }
仔细想想，如果线程 a 先获取了读锁，然后获取写锁，那么线程 a 就到阻塞队列休眠了，自己把自己弄休眠了，而且可能之后就没人去唤醒它了。
总结
写锁的获取 ：只要有读或写锁被占用，就排队
读锁的获取 ：有其他线程占用写锁就排队。公平模式下，先来先到的排队方式；非公平模式下，可以和第一个竞争。抢不到，排队。
同时state 分成两部分，所以就是最大65535 个线程同时等待锁。
疑问
- tryAcquireShared 方法内 rh.count++; 要是此时的情况是 AB 然后 有个 B ，C 的线程过来同时获取锁，CAS成功后先后进入rh.count ++ 这一行，那么这样不会产生并发问题了吗？
参考资料 ：
原文地址：https://www.cnblogs.com/Benjious/p/10161647.html
时间： 2024-10-11 03:09:35