java 并发(五)---AbstractQueuedSynchronizer（4）-白红宇

java 并发(五)---AbstractQueuedSynchronizer（4）

阅读量：7079 次

发布时间：2019-06-28

本文共 17871 字，大约阅读时间需要 59 分钟。

问题：

rwl 的底层实现是什么，应用场景是什么

读写锁 ReentrantReadWriteLock

首先我们来了解一下 ReentrantReadWriteLock 的作用是什么？和 ReentranLock 有什么区别？Reentrancy 英文的意思是可重入性。ReentrantReadWriteLock下文简称（rrwl）

下面总结来自，你也可以从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
     
       m = new TreeMap
      
       ();  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
     
       { 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 }

1 protected final int tryAcquireShared(int unused) {  2   3     Thread current = Thread.currentThread();  4     int c = getState();  5   6     // exclusiveCount(c) 不等于 0，说明有线程持有写锁，  7     //    而且不是当前线程持有写锁，那么当前线程获取读锁失败  8     //         （另，如果持有写锁的是当前线程，是可以继续获取读锁的）  9     if (exclusiveCount(c) != 0 && 10         getExclusiveOwnerThread() != current) 11         return -1; 12  13     // 读锁的获取次数 14     int r = sharedCount(c); 15  16     // 读锁获取是否需要被阻塞，稍后细说。为了进去下面的分支，假设这里不阻塞就好了 17     if (!readerShouldBlock() && 18         // 判断是否会溢出 (2^16-1，没那么容易溢出的) 19         r < MAX_COUNT && 20         // 下面这行 CAS 是将 state 属性的高 16 位加 1，低 16 位不变，如果成功就代表获取到了读锁 21         compareAndSetState(c, c + SHARED_UNIT)) { 22  23         // ======================= 24         //   进到这里就是获取到了读锁 25         // ======================= 26  27         if (r == 0) { 28             // r == 0 说明此线程是第一个获取读锁的，或者说在它前面获取读锁的都走光光了，它也算是第一个吧 29             //  记录 firstReader 为当前线程，及其持有的读锁数量：1 30             firstReader = current; 31             firstReaderHoldCount = 1; 32         } else if (firstReader == current) { 33             // 进来这里，说明是 firstReader 重入获取读锁（这非常简单，count 加 1 结束） 34             firstReaderHoldCount++; 35         } else { 36             // 前面我们说了 cachedHoldCounter 用于缓存最后一个获取读锁的线程 37             // 如果 cachedHoldCounter 缓存的不是当前线程，设置为缓存当前线程的 HoldCounter 38             HoldCounter rh = cachedHoldCounter; 39             if (rh == null || rh.tid != getThreadId(current)) 40                 cachedHoldCounter = rh = readHolds.get(); 41             else if (rh.count == 0) 42                 // 到这里，那么就是 cachedHoldCounter 缓存的是当前线程，但是 count 为 0， 43                 // 大家可以思考一下：这里为什么要 set ThreadLocal 呢？(当然，答案肯定不在这块代码中) 44                 //   既然 cachedHoldCounter 缓存的是当前线程， 45                 //   当前线程肯定调用过 readHolds.get() 进行初始化 ThreadLocal 46                 readHolds.set(rh); 47  48             // count 加 1 49             rh.count++; 50         } 51         // return 大于 0 的数，代表获取到了共享锁 52         return 1; 53     } 54     // 往下看 55     return fullTryAcquireShared(current); 56 }

上面的代码中，要进入 if 分支，需要满足：readerShouldBlock() 返回 false，并且 CAS 要成功（我们先不要纠结 MAX_COUNT 溢出。那我们反向推，怎么样进入到最后的 fullTryAcquireShared ：

readerShouldBlock() 返回 true，2 种情况：
- 在 FairSync 中说的是 hasQueuedPredecessors()，即阻塞队列中有其他元素在等待锁。
  
  也就是说，公平模式下，有人在排队呢，你新来的不能直接获取锁
- 在 NonFairSync 中说的是 apparentlyFirstQueuedIsExclusive()，即判断阻塞队列中 head 的第一个后继节点是否是来获取写锁的，如果是的话，让这个写锁先来，避免写锁饥饿。
  
  作者给写锁定义了更高的优先级，所以如果碰上获取写锁的线程马上就要获取到锁了，获取读锁的线程不应该和它抢。
  
  如果 head.next 不是来获取写锁的，那么可以随便抢，因为是非公平模式，大家比比 CAS 速度

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 }

读完这一段，我觉得有两个问题值得我们思考：

问题一：

上面的代码还有一句让我们值的思考的是

rh.count++

这是在多线程环境下,那这里会不会引起线程安全问题呢?不会,原因rh 来到这里会去ThreadLocal 拿一个值，所以不存在线程问题。

问题二：

HoldCounter rh = cachedHoldCounter;if (rh == null || rh.tid != getThreadId(current))      cachedHoldCounter = rh = readHolds.get();else if (rh.count == 0)       readHolds.set(rh);rh.count++;

else-if 里的条件什么时候满足呢？可以理解为一个线程先lock 再unlock 又再次lock 的场景。把第一次unlock执行的local执行的remove对象再补充回去的情况。

读锁释放

下面我们看看读锁释放的流程：

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 先获取了读锁，然后获取写锁，（tryAcquire方法返回false）那么线程 a 就到阻塞队列休眠了，自己把自己弄休眠了，而且可能之后就没人去唤醒它了。