从源码深入理解读写锁(golang-RWMutex)

╰半橙微兮° 2024-03-22 13:16 37阅读 0赞

在读多写少的情况下，即使一段时间内没有写操作，大量并发的读访问也不得不在Mutex的保护下变成串行访问，这种情况下，使用Mutex，对性能影响比较大。  
所以就要区分读写操作。如果某个读操作的g持有了锁，其他读操作的g就不必等待了，可以并发的访问共享变量，这样就可以将串行的读变成并行的读，提高读操作的性能。**可理解为共享锁。**  
当写操作的g持有锁，它是一个**排他锁**，不管其他的g是写操作还是读操作，都需要阻塞等待持有锁的g释放锁。

## 什么是RWMutex？ ##

reader/writer互斥锁，在某一时刻只能由任意数量的reader持有，或者是只被单个writer持有。  
RWMutex实现了5个方法：

*  Lock/Unlock：写操作时调用。如果锁已经被reader或者writer持有，那么，Lock方法会一直阻塞，直到能获取到锁；Unlock是对应的释放锁方法
 *  RLock/RUnlock：读操作时调用。如果锁已经被writer持有，RLock方法会一直阻塞，直到能获取锁，否则直接return；Rnlock是对应的释放锁方法
 *  RLocker：这个方法的作用是为读操作返回一个 Locker 接口的对象

案例：计数器，1writer n reader

## 使用场景 ##

**如果可以明确区分 reader 和 writer goroutine ，且有大量的并发读，少量的并发写，并且有强烈的性能要求，可以考虑使用读写锁RWMutex替换Mutex**

## 实现原理 ##

RWMutex 是很常见的并发原语，很多编程语言的库都提供了类似的并发类型。RWMutex  
一般都是基于互斥锁、条件变量（condition variables）或者信号量（semaphores）等  
并发原语来实现。**Go 标准库中的 RWMutex 是基于 Mutex 实现的。**  
**reader-writers 问题，一般有三类，基于对读和写操作的优先级，读写锁的设计和实现也分成三类**

*  **Read-Preferring：读优先的设计可以提供很高的并发性。**但在竞争激烈的情况下会导致写饥饿
 *  **Write-Preferring**：如果有一个writer在等待请求锁，它会阻止新来请求锁reader获取到锁，优先保障writer。当然，如果reader已经获得锁，新请求的writer也需要等待已持有锁的reader释放锁。写优先级设计中的优先权是针对新来的请求而言的。这种设计主要避免了 writer 的饥饿问题。
 *  **不指定优先级**：这种设计比较简单，不区分 reader 和 writer 优先级，某些场景下这种不指定优先级的设计反而更有效，因为第一类优先级会导致写饥饿，第二类优先级可能会导致读饥饿，这种不指定优先级的访问不再区分读写，大家都是同一个优先级，解决了饥饿的问题。

**Go 标准库中的 RWMutex 设计是 Write-preferring 方案。一个正在阻塞的 Lock 调用**  
**会排除新的 reader 请求到锁。**

## 源码解析 ##

上锁解锁流程以及数值变化情况

![163f5b45276370bec965f3ba14598d47.jpeg][]

rwmutexMaxReaders 的数量被初始化为`1<<30`，理想中，写锁不会持续很久，不会导致readerCount 自动从负值自动+1回到正值。

### RLock/RUnlock实现[\#][Link 1] ###

Copy

`type RWMutex struct { w sync.Mutex // hold if there are pending writers writerSem uint32 // 写 阻塞信号 readerSem uint32 // 读 阻塞信号 readerCount int32 // 正在读的调用者数量/ 当为负数时 表示有write持有锁 readerWait int32 // writer持有锁之前正等待解锁的数量 } const rwmutexMaxReaders = 1 << 30 func (rw *RWMutex) RLock() { if atomic.AddInt32(&rw.readerCount, 1) < 0 { // 写端 持有锁， 读端阻塞 runtime_SemacquireMutex(&rw.readerSem, false, 0) } } func (rw *RWMutex) RUnlock() { if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 { rw.rUnlockSlow(r) } } func (rw *RWMutex) rUnlockSlow(r int32) { if r+1 == 0 || r+1 == -rwmutexMaxReaders { fatal("sync: RUnlock of unlocked RWMutex") } if atomic.AddInt32(&rw.readerWait, -1) == 0 { // 无读者等待，唤醒写端等待者 runtime_Semrelease(&rw.writerSem, false, 1) } }`

#### RLock[\#][Link 2] ####

第11行，上读锁，首先对readerCount进行原子加1，如果小于0则表示存在写锁，直接阻塞。为什么readerCount会存在负值？这个要看readerCount除了在RLock中处理，还在哪里被处理了。可以看到在获取写锁时有响应代码。后面在解释。如果原子加大于等于0，则表示获取读锁成功。

![4937bf1b84ee26dc7d35d484c3e014af.jpeg][]

#### RUnlock[\#][Link 3] ####

第18行，读解锁，对readerCount进行原子减1，如果小于零，则表示存在活跃的reader(即当前获得互斥锁的写锁之前获取到读锁权限的读者数量)，readerWait 字段就减 1，直到所有的活跃的 reader 都释放了读锁，才会唤醒这个 write

![f50fbdb293e6e6abdb705a65828057c5.jpeg][]

### Lock/Unlock[\#][Link 4] ###

Copy

`func (rw *RWMutex) Lock() { // 1. 先尝试获取互斥锁 rw.w.Lock() // 2. 看是否有其他正持有锁的读者，有则阻塞 r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 { // rc - rwmutexMaxReaders + rwmutexMaxReaders > 0说明还有等待者, 写端阻塞 runtime_SemacquireMutex(&rw.writerSem, false, 0) } } func (rw *rwMutex) Unlock() { r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders) if r >= rwmutexMaxReaders { fatal("sync: Unlock of unlocked RWMutex") } // 如果有等待的读者，先唤醒 for i := 0; i < int(r); i++ { runtime_Semrelease(&rw.readerSem, false, 0) } // 释放互斥锁 rw.w.Unlock() }`

#### Lock[\#][Link 5] ####

1.  先获取互斥锁
2.  成功获取后，r=readerCount-rwmutexMaxReaders，得到的数值就是一个负数，在加上rwmutexReaders就**表示写锁等待者的数量**，此时，如果r不等于0，且readerWait+r!=0，则表示有读等待者，**写锁阻塞**

我们知道，写操作要等待读操作结束后才可以获得锁，写操作等待期间可能还有新的读操作持续到来，如果写操作等待所有读操作结束，就会出现饥饿现象。然而，通过`readerWait`可完美解决这个问题。

写操作到来时，会把`readerCount`值拷贝到`readerWait`中，用于标记排在写操作之前到读者个数。  
当读操作结束后，除了会递减`readerCount`，还会递减`readerWait`的值，当`readerWait`值变为0时会唤醒写操作。

写操作之后产生的读操作会加入到`readerCount`中，阻塞知道写锁释放。

![41ed2a688cb100ca35ce78a28303fe3c.jpeg][]

#### Unlock[\#][Link 6] ####

上面说过，写锁之后来的读者会被阻塞，所以在写锁释放之际，会看是否有需要唤醒的读者，再释放互斥锁

![da80342efc4251a5e2aa347a4e0a375e.jpeg][]

## 场景讨论 ##

### 写操作如何阻塞写操作[\#][Link 7] ###

读写锁包含一个互斥锁(Mutex)，写锁必须先获取该互斥锁，如果互斥锁已被协程A获取，意味者其他协程只能阻塞等待互斥锁释放

### 写操作是如何阻塞读操作[\#][Link 8] ###

`readerCount`是个整型值，用于表示读者数量，不考虑写操作的情况下，每次获取读锁，将该值加1，每次解锁将其减1，所以`readerCount`的取值为`[0, N]`，最大可支持`2^30`个并发读者。

当写锁定进行时，会先将`readerCount -= rwmutextMaxReaders(2^30)`，此时 `readerCount`负数。这时再有读者到了，检测到`readerCount`为负值，则表示有写操作正在进行，后来到读者阻塞等待。等待者的数量即 `reaerCount + 2^30`

### 读操作是如何阻止写操作的[\#][Link 9] ###

写操作时，会把`readerCount`的值拷贝到`readerWait`中，用于标记在写操作前面读者的个数，前面的写锁释放后，会递减`readerCount，readerWait`，当`readerWait`值变为0时唤醒写操作

## 3个踩坑点 ##

### 不可复制[\#][Link 10] ###

rwmutex是由一个互斥锁和四个辅助字段组成的，与互斥锁一样，读写锁也是不能复制的。  
一旦读写锁被使用，它的字段就会记录它当前的一些状态，如果此时去复制这把锁，就会把它的状态也复制过去。但原来的锁在释放的时候，并不会修改复制出来的读写锁，会导致复制出来的读写锁状态异常，可能永远无法释放锁。

### 重入导致死锁[\#][Link 11] ###

读写锁重入，或者递归调用，导致的死锁情况很多

1.  读写锁内部基于互斥锁实现对writer并发控制，而互斥锁本身就有重入问题，所以，writer重入调用Lock，会导致死锁

Copy

`func foo(l *sync.RWMutex) { fmt.Println("lock in foo") l.Lock() bar(l) l.Unlock() } func bar(l *sync.RWMutex) { fmt.Println("lock in bar") l.Lock() l.Unlock() } func main() { l := &sync.RWMutex{} foo(l) }`

1.  当一个 writer 请求锁的时候，如果已经有一些活跃的 reader，它会等待这些活跃的reader 完成，才有可能获取到锁，但是，如果之后活跃的 reader 再依赖新的 reader 的话，这些新的 reader 就会等待 writer 释放锁之后才能继续执行，这就形成了一个环形依赖： writer 依赖活跃的 reader -> 活跃的 reader 依赖新来的 reader -> 新来的 reader依赖 writer。

Copy

`func main() { var mu sync.RWMutex go func() { time.Sleep(200*time.Millisecond) mu.Lock() fmt.Println("Lock") time.Sleep(100*time.Millisend) mu.Unlock() fmt.Println("Unlock") } go func() { factorial(&mu, 10) // 计算10的阶乘 } select {} } // func factorial(m *sync.RWMutex, n int) { if n < 1 { return 0 } fmt.Println("RLock") m.RLock() defer func() { fmt.Println("RUnlock") m.RUnlock() } time.Sleep(100*time.Millisecond) return factorial(m, n-1) * n }`

factorial 方法是一个递归计算阶乘的方法，我们用它来模拟 reader。为了更容易地制造出死锁场景，在这里加上了 sleep 的调用，延缓逻辑的执行。这个方法会调用读锁（第 27  
行），在第 33 行递归地调用此方法，每次调用都会产生一次读锁的调用，所以可以不断地产生读锁的调用，而且必须等到新请求的读锁释放，这个读锁才能释放。同时，我们使用另一个 goroutine 去调用 Lock 方法，来实现 writer，这个 writer 会等待200 毫秒后才会调用 Lock，这样在调用 Lock 的时候，factoria 方法还在执行中不断调用  
RLock。这两个 goroutine 互相持有锁并等待，谁也不会退让一步，满足了“writer 依赖活跃的reader -> 活跃的 reader 依赖新来的 reader -> 新来的 reader 依赖 writer”的死锁条件，所以就导致了死锁的产生。

### 释放未加锁的RWMutex[\#][Link 12] ###

锁都是成对出现的，Lock和RLock的多余调用会导致锁没有被释放，可能会出现死锁。  
而Unlock和RUnlock多余调用会导致panic

[163f5b45276370bec965f3ba14598d47.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/83c051306e924d4dbeb614c0d6338eff.png
[Link 1]: https://www.cnblogs.com/panlq/p/17373428.html#2381190812
[Link 2]: https://www.cnblogs.com/panlq/p/17373428.html#3145477331
[4937bf1b84ee26dc7d35d484c3e014af.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/dc115a4f9e9046fe8599a17ac874699a.png
[Link 3]: https://www.cnblogs.com/panlq/p/17373428.html#2498196746
[f50fbdb293e6e6abdb705a65828057c5.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/1d74aa2279144df4acab415db4c84d0d.png
[Link 4]: https://www.cnblogs.com/panlq/p/17373428.html#3933432459
[Link 5]: https://www.cnblogs.com/panlq/p/17373428.html#4079340613
[41ed2a688cb100ca35ce78a28303fe3c.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/4b939d538e00464f88bbb4559a28ab30.png
[Link 6]: https://www.cnblogs.com/panlq/p/17373428.html#933578552
[da80342efc4251a5e2aa347a4e0a375e.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/f19c946dd1c846ceb626d3414dd3063e.png
[Link 7]: https://www.cnblogs.com/panlq/p/17373428.html#4162869748
[Link 8]: https://www.cnblogs.com/panlq/p/17373428.html#3130252114
[Link 9]: https://www.cnblogs.com/panlq/p/17373428.html#114757893
[Link 10]: https://www.cnblogs.com/panlq/p/17373428.html#1977518214
[Link 11]: https://www.cnblogs.com/panlq/p/17373428.html#2034860272
[Link 12]: https://www.cnblogs.com/panlq/p/17373428.html#2562748666