|
| 1 | + |
| 2 | + |
| 3 | +# 1.读写锁的介绍 # |
| 4 | +在并发场景中用于解决线程安全的问题,我们几乎会高频率的使用到独占式锁,通常使用java提供的关键字synchronized(关于synchronized可以[看这篇文章](https://juejin.im/post/5ae6dc04f265da0ba351d3ff))或者concurrents包中实现了Lock接口的[ReentrantLock](https://juejin.im/post/5aeb0a8b518825673a2066f0)。它们都是独占式获取锁,也就是在同一时刻只有一个线程能够获取锁。而在一些业务场景中,大部分只是读数据,写数据很少,如果仅仅是读数据的话并不会影响数据正确性(出现脏读),而如果在这种业务场景下,依然使用独占锁的话,很显然这将是出现性能瓶颈的地方。针对这种读多写少的情况,java还提供了另外一个实现Lock接口的ReentrantReadWriteLock(读写锁)。**读写所允许同一时刻被多个读线程访问,但是在写线程访问时,所有的读线程和其他的写线程都会被阻塞**。在分析WirteLock和ReadLock的互斥性时可以按照WriteLock与WriteLock之间,WriteLock与ReadLock之间以及ReadLock与ReadLock之间进行分析。更多关于读写锁特性介绍大家可以看源码上的介绍(阅读源码时最好的一种学习方式,我也正在学习中,与大家共勉),这里做一个归纳总结: |
| 5 | + |
| 6 | +1. **公平性选择**:支持非公平性(默认)和公平的锁获取方式,吞吐量还是非公平优于公平; |
| 7 | +2. **重入性**:支持重入,读锁获取后能再次获取,写锁获取之后能够再次获取写锁,同时也能够获取读锁; |
| 8 | +3. **锁降级**:遵循获取写锁,获取读锁再释放写锁的次序,写锁能够降级成为读锁 |
| 9 | + |
| 10 | +要想能够彻底的理解读写锁必须能够理解这样几个问题:1. 读写锁是怎样实现分别记录读写状态的?2. 写锁是怎样获取和释放的?3.读锁是怎样获取和释放的?我们带着这样的三个问题,再去了解下读写锁。 |
| 11 | + |
| 12 | +# 2.写锁详解 # |
| 13 | + |
| 14 | +## 2.1.写锁的获取 ## |
| 15 | +同步组件的实现聚合了同步器(AQS),并通过重写重写同步器(AQS)中的方法实现同步组件的同步语义(关于同步组件的实现层级结构可以[看这篇文章](https://juejin.im/post/5aeb055b6fb9a07abf725c8c),AQS的底层实现分析可以[看这篇文章](https://juejin.im/post/5aeb07ab6fb9a07ac36350c8))。因此,写锁的实现依然也是采用这种方式。在同一时刻写锁是不能被多个线程所获取,很显然写锁是独占式锁,而实现写锁的同步语义是通过重写AQS中的tryAcquire方法实现的。源码为: |
| 16 | + |
| 17 | + protected final boolean tryAcquire(int acquires) { |
| 18 | + /* |
| 19 | + * Walkthrough: |
| 20 | + * 1. If read count nonzero or write count nonzero |
| 21 | + * and owner is a different thread, fail. |
| 22 | + * 2. If count would saturate, fail. (This can only |
| 23 | + * happen if count is already nonzero.) |
| 24 | + * 3. Otherwise, this thread is eligible for lock if |
| 25 | + * it is either a reentrant acquire or |
| 26 | + * queue policy allows it. If so, update state |
| 27 | + * and set owner. |
| 28 | + */ |
| 29 | + Thread current = Thread.currentThread(); |
| 30 | + // 1. 获取写锁当前的同步状态 |
| 31 | + int c = getState(); |
| 32 | + // 2. 获取写锁获取的次数 |
| 33 | + int w = exclusiveCount(c); |
| 34 | + if (c != 0) { |
| 35 | + // (Note: if c != 0 and w == 0 then shared count != 0) |
| 36 | + // 3.1 当读锁已被读线程获取或者当前线程不是已经获取写锁的线程的话 |
| 37 | + // 当前线程获取写锁失败 |
| 38 | + if (w == 0 || current != getExclusiveOwnerThread()) |
| 39 | + return false; |
| 40 | + if (w + exclusiveCount(acquires) > MAX_COUNT) |
| 41 | + throw new Error("Maximum lock count exceeded"); |
| 42 | + // Reentrant acquire |
| 43 | + // 3.2 当前线程获取写锁,支持可重复加锁 |
| 44 | + setState(c + acquires); |
| 45 | + return true; |
| 46 | + } |
| 47 | + // 3.3 写锁未被任何线程获取,当前线程可获取写锁 |
| 48 | + if (writerShouldBlock() || |
| 49 | + !compareAndSetState(c, c + acquires)) |
| 50 | + return false; |
| 51 | + setExclusiveOwnerThread(current); |
| 52 | + return true; |
| 53 | + } |
| 54 | + |
| 55 | +这段代码的逻辑请看注释,这里有一个地方需要重点关注,exclusiveCount(c)方法,该方法源码为: |
| 56 | + |
| 57 | + static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; } |
| 58 | +其中**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位用来表示写锁的获取次数**。同时还有一个方法值得我们注意: |
| 59 | + |
| 60 | + static int sharedCount(int c) { return c >>> SHARED_SHIFT; } |
| 61 | + |
| 62 | +该方法是获取读锁被获取的次数,是将同步状态(int c)右移16次,即取同步状态的高16位,现在我们可以得出另外一个结论**同步状态的高16位用来表示读锁被获取的次数**。现在还记得我们开篇说的需要弄懂的第一个问题吗?读写锁是怎样实现分别记录读锁和写锁的状态的,现在这个问题的答案就已经被我们弄清楚了,其示意图如下图所示: |
| 63 | + |
| 64 | + |
| 65 | + |
| 66 | + |
| 67 | +现在我们回过头来看写锁获取方法tryAcquire,其主要逻辑为:**当读锁已经被读线程获取或者写锁已经被其他写线程获取,则写锁获取失败;否则,获取成功并支持重入,增加写状态。** |
| 68 | + |
| 69 | +## 2.2.写锁的释放 ## |
| 70 | +写锁释放通过重写AQS的tryRelease方法,源码为: |
| 71 | + |
| 72 | + protected final boolean tryRelease(int releases) { |
| 73 | + if (!isHeldExclusively()) |
| 74 | + throw new IllegalMonitorStateException(); |
| 75 | + //1. 同步状态减去写状态 |
| 76 | + int nextc = getState() - releases; |
| 77 | + //2. 当前写状态是否为0,为0则释放写锁 |
| 78 | + boolean free = exclusiveCount(nextc) == 0; |
| 79 | + if (free) |
| 80 | + setExclusiveOwnerThread(null); |
| 81 | + //3. 不为0则更新同步状态 |
| 82 | + setState(nextc); |
| 83 | + return free; |
| 84 | + } |
| 85 | + |
| 86 | +源码的实现逻辑请看注释,不难理解与ReentrantLock基本一致,这里需要注意的是,减少写状态` int nextc = getState() - releases;`只需要用**当前同步状态直接减去写状态的原因正是我们刚才所说的写状态是由同步状态的低16位表示的**。 |
| 87 | + |
| 88 | +# 3.读锁详解 # |
| 89 | +## 3.1.读锁的获取 ## |
| 90 | +看完了写锁,现在来看看读锁,读锁不是独占式锁,即同一时刻该锁可以被多个读线程获取也就是一种共享式锁。按照之前对AQS介绍,实现共享式同步组件的同步语义需要通过重写AQS的tryAcquireShared方法和tryReleaseShared方法。读锁的获取实现方法为: |
| 91 | + |
| 92 | + protected final int tryAcquireShared(int unused) { |
| 93 | + /* |
| 94 | + * Walkthrough: |
| 95 | + * 1. If write lock held by another thread, fail. |
| 96 | + * 2. Otherwise, this thread is eligible for |
| 97 | + * lock wrt state, so ask if it should block |
| 98 | + * because of queue policy. If not, try |
| 99 | + * to grant by CASing state and updating count. |
| 100 | + * Note that step does not check for reentrant |
| 101 | + * acquires, which is postponed to full version |
| 102 | + * to avoid having to check hold count in |
| 103 | + * the more typical non-reentrant case. |
| 104 | + * 3. If step 2 fails either because thread |
| 105 | + * apparently not eligible or CAS fails or count |
| 106 | + * saturated, chain to version with full retry loop. |
| 107 | + */ |
| 108 | + Thread current = Thread.currentThread(); |
| 109 | + int c = getState(); |
| 110 | + //1. 如果写锁已经被获取并且获取写锁的线程不是当前线程的话,当前 |
| 111 | + // 线程获取读锁失败返回-1 |
| 112 | + if (exclusiveCount(c) != 0 && |
| 113 | + getExclusiveOwnerThread() != current) |
| 114 | + return -1; |
| 115 | + int r = sharedCount(c); |
| 116 | + if (!readerShouldBlock() && |
| 117 | + r < MAX_COUNT && |
| 118 | + //2. 当前线程获取读锁 |
| 119 | + compareAndSetState(c, c + SHARED_UNIT)) { |
| 120 | + //3. 下面的代码主要是新增的一些功能,比如getReadHoldCount()方法 |
| 121 | + //返回当前获取读锁的次数 |
| 122 | + if (r == 0) { |
| 123 | + firstReader = current; |
| 124 | + firstReaderHoldCount = 1; |
| 125 | + } else if (firstReader == current) { |
| 126 | + firstReaderHoldCount++; |
| 127 | + } else { |
| 128 | + HoldCounter rh = cachedHoldCounter; |
| 129 | + if (rh == null || rh.tid != getThreadId(current)) |
| 130 | + cachedHoldCounter = rh = readHolds.get(); |
| 131 | + else if (rh.count == 0) |
| 132 | + readHolds.set(rh); |
| 133 | + rh.count++; |
| 134 | + } |
| 135 | + return 1; |
| 136 | + } |
| 137 | + //4. 处理在第二步中CAS操作失败的自旋已经实现重入性 |
| 138 | + return fullTryAcquireShared(current); |
| 139 | + } |
| 140 | + |
| 141 | +代码的逻辑请看注释,需要注意的是 **当写锁被其他线程获取后,读锁获取失败**,否则获取成功利用CAS更新同步状态。另外,当前同步状态需要加上SHARED_UNIT(`(1 << SHARED_SHIFT)`即0x00010000)的原因这是我们在上面所说的同步状态的高16位用来表示读锁被获取的次数。如果CAS失败或者已经获取读锁的线程再次获取读锁时,是靠fullTryAcquireShared方法实现的,这段代码就不展开说了,有兴趣可以看看。 |
| 142 | + |
| 143 | +## 3.2.读锁的释放 ## |
| 144 | +读锁释放的实现主要通过方法tryReleaseShared,源码如下,主要逻辑请看注释: |
| 145 | + |
| 146 | + protected final boolean tryReleaseShared(int unused) { |
| 147 | + Thread current = Thread.currentThread(); |
| 148 | + // 前面还是为了实现getReadHoldCount等新功能 |
| 149 | + if (firstReader == current) { |
| 150 | + // assert firstReaderHoldCount > 0; |
| 151 | + if (firstReaderHoldCount == 1) |
| 152 | + firstReader = null; |
| 153 | + else |
| 154 | + firstReaderHoldCount--; |
| 155 | + } else { |
| 156 | + HoldCounter rh = cachedHoldCounter; |
| 157 | + if (rh == null || rh.tid != getThreadId(current)) |
| 158 | + rh = readHolds.get(); |
| 159 | + int count = rh.count; |
| 160 | + if (count <= 1) { |
| 161 | + readHolds.remove(); |
| 162 | + if (count <= 0) |
| 163 | + throw unmatchedUnlockException(); |
| 164 | + } |
| 165 | + --rh.count; |
| 166 | + } |
| 167 | + for (;;) { |
| 168 | + int c = getState(); |
| 169 | + // 读锁释放 将同步状态减去读状态即可 |
| 170 | + int nextc = c - SHARED_UNIT; |
| 171 | + if (compareAndSetState(c, nextc)) |
| 172 | + // Releasing the read lock has no effect on readers, |
| 173 | + // but it may allow waiting writers to proceed if |
| 174 | + // both read and write locks are now free. |
| 175 | + return nextc == 0; |
| 176 | + } |
| 177 | + } |
| 178 | + |
| 179 | +# 4.锁降级 # |
| 180 | +读写锁支持锁降级,**遵循按照获取写锁,获取读锁再释放写锁的次序,写锁能够降级成为读锁**,不支持锁升级,关于锁降级下面的示例代码摘自ReentrantWriteReadLock源码中: |
| 181 | + |
| 182 | + void processCachedData() { |
| 183 | + rwl.readLock().lock(); |
| 184 | + if (!cacheValid) { |
| 185 | + // Must release read lock before acquiring write lock |
| 186 | + rwl.readLock().unlock(); |
| 187 | + rwl.writeLock().lock(); |
| 188 | + try { |
| 189 | + // Recheck state because another thread might have |
| 190 | + // acquired write lock and changed state before we did. |
| 191 | + if (!cacheValid) { |
| 192 | + data = ... |
| 193 | + cacheValid = true; |
| 194 | + } |
| 195 | + // Downgrade by acquiring read lock before releasing write lock |
| 196 | + rwl.readLock().lock(); |
| 197 | + } finally { |
| 198 | + rwl.writeLock().unlock(); // Unlock write, still hold read |
| 199 | + } |
| 200 | + } |
| 201 | + |
| 202 | + try { |
| 203 | + use(data); |
| 204 | + } finally { |
| 205 | + rwl.readLock().unlock(); |
| 206 | + } |
| 207 | + } |
| 208 | + } |
0 commit comments