你好，我是鸟窝。
在前面的四节课中，我们学习了第一个同步原语，即Mutex，我们使用它来保证读写共享资源的安全性。不管是读还是写，我们都通过Mutex来保证只有一个goroutine访问共享资源，这在某些情况下有点“浪费”。比如说，在写少读多的情况下，即使一段时间内没有写操作，大量并发的读访问也不得不在Mutex的保护下变成了串行访问，这个时候，使用Mutex，对性能的影响就比较大。
怎么办呢？你是不是已经有思路了，对，就是区分读写操作。
我来具体解释一下。如果某个读操作的goroutine持有了锁，在这种情况下，其它读操作的goroutine就不必一直傻傻地等待了，而是可以并发地访问共享变量，这样我们就可以将串行的读变成并行读，提高读操作的性能。当写操作的goroutine持有锁的时候，它就是一个排外锁，其它的写操作和读操作的goroutine，需要阻塞等待持有这个锁的goroutine释放锁。
这一类并发读写问题叫作readers-writers问题，意思就是，同时可能有多个读或者多个写，但是只要有一个线程在执行写操作，其它的线程都不能执行读写操作。
Go标准库中的RWMutex（读写锁）就是用来解决这类readers-writers问题的。所以，这节课，我们就一起来学习RWMutex。我会给你介绍读写锁的使用场景、实现原理以及容易掉入的坑，你一定要记住这些陷阱，避免在实际的开发中犯相同的错误。
什么是RWMutex？
我先简单解释一下读写锁RWMutex。标准库中的RWMutex是一个 reader/writer 互斥锁。RWMutex在某一时刻只能由任意数量的reader持有，或者是只被单个的writer持有。
RWMutex的方法也很少，总共有5个。
- Lock/Unlock：写操作时调用的方法。如果锁已经被reader或者writer持有，那么，Lock方法会一直阻塞，直到能获取到锁；Unlock则是配对的释放锁的方法。
- RLock/RUnlock：读操作时调用的方法。如果锁已经被writer持有的话，RLock方法会一直阻塞，直到能获取到锁，否则就直接返回；而RUnlock是reader释放锁的方法。
- RLocker：这个方法的作用是为读操作返回一个Locker接口的对象。它的Lock方法会调用RWMutex的RLock方法，它的Unlock方法会调用RWMutex的RUnlock方法。
RWMutex的零值是未加锁的状态，所以，当你使用RWMutex的时候，无论是声明变量，还是嵌入到其它struct中，都不必显式地初始化。
我以计数器为例，来说明一下，如何使用RWMutex保护共享资源。计数器的count++操作是写操作，而获取count的值是读操作，这个场景非常适合读写锁，因为读操作可以并行执行，写操作时只允许一个线程执行，这正是readers-writers问题。
在这个例子中，使用10个goroutine进行读操作，每读取一次，sleep 1毫秒，同时，还有一个gorotine进行写操作，每一秒写一次，这是一个 1 writer-n reader 的读写场景，而且写操作还不是很频繁（一秒一次）：
func main() {
var counter Counter
for i := 0; i < 10; i++ { // 10个reader
go func() {
for {
counter.Count() // 计数器读操作
time.Sleep(time.Millisecond)
}
}()
}
for { // 一个writer
counter.Incr() // 计数器写操作
time.Sleep(time.Second)
}
}
// 一个线程安全的计数器
type Counter struct {
mu sync.RWMutex
count uint64
}
// 使用写锁保护
func (c *Counter) Incr() {
c.mu.Lock()
c.count++
c.mu.Unlock()
}
// 使用读锁保护
func (c *Counter) Count() uint64 {
c.mu.RLock()
defer c.mu.RUnlock()
return c.count
}
可以看到，Incr方法会修改计数器的值，是一个写操作，我们使用Lock/Unlock进行保护。Count方法会读取当前计数器的值，是一个读操作，我们使用RLock/RUnlock方法进行保护。
Incr方法每秒才调用一次，所以，writer竞争锁的频次是比较低的，而10个goroutine每毫秒都要执行一次查询，通过读写锁，可以极大提升计数器的性能，因为在读取的时候，可以并发进行。如果使用Mutex，性能就不会像读写锁这么好。因为多个reader并发读的时候，使用互斥锁导致了reader要排队读的情况，没有RWMutex并发读的性能好。
如果你遇到可以明确区分reader和writer goroutine的场景，且有大量的并发读、少量的并发写，并且有强烈的性能需求，你就可以考虑使用读写锁RWMutex替换Mutex。
在实际使用RWMutex的时候，如果我们在struct中使用RWMutex保护某个字段，一般会把它和这个字段放在一起，用来指示两个字段是一组字段。除此之外，我们还可以采用匿名字段的方式嵌入struct，这样，在使用这个struct时，我们就可以直接调用Lock/Unlock、RLock/RUnlock方法了，这和我们前面在01讲中介绍Mutex的使用方法很类似，你可以回去复习一下。
RWMutex的实现原理
RWMutex是很常见的并发原语，很多编程语言的库都提供了类似的并发类型。RWMutex一般都是基于互斥锁、条件变量（condition variables）或者信号量（semaphores）等并发原语来实现。Go标准库中的RWMutex是基于Mutex实现的。
readers-writers问题一般有三类，基于对读和写操作的优先级，读写锁的设计和实现也分成三类。
- Read-preferring：读优先的设计可以提供很高的并发性，但是，在竞争激烈的情况下可能会导致写饥饿。这是因为，如果有大量的读，这种设计会导致只有所有的读都释放了锁之后，写才可能获取到锁。
- Write-preferring：写优先的设计意味着，如果已经有一个writer在等待请求锁的话，它会阻止新来的请求锁的reader获取到锁，所以优先保障writer。当然，如果有一些reader已经请求了锁的话，新请求的writer也会等待已经存在的reader都释放锁之后才能获取。所以，写优先级设计中的优先权是针对新来的请求而言的。这种设计主要避免了writer的饥饿问题。
- 不指定优先级：这种设计比较简单，不区分reader和writer优先级，某些场景下这种不指定优先级的设计反而更有效，因为第一类优先级会导致写饥饿，第二类优先级可能会导致读饥饿，这种不指定优先级的访问不再区分读写，大家都是同一个优先级，解决了饥饿的问题。
Go标准库中的RWMutex设计是Write-preferring方案。一个正在阻塞的Lock调用会排除新的reader请求到锁。
RWMutex包含一个Mutex，以及四个辅助字段writerSem、readerSem、readerCount和readerWait：
type RWMutex struct {
w Mutex // 互斥锁解决多个writer的竞争
writerSem uint32 // writer信号量
readerSem uint32 // reader信号量
readerCount int32 // reader的数量
readerWait int32 // writer等待完成的reader的数量
}
const rwmutexMaxReaders = 1 << 30
我来简单解释一下这几个字段。
- 字段w：为writer的竞争锁而设计；
- 字段readerCount：记录当前reader的数量（以及是否有writer竞争锁）；
- readerWait：记录writer请求锁时需要等待read完成的reader的数量；
- writerSem 和readerSem：都是为了阻塞设计的信号量。
这里的常量rwmutexMaxReaders，定义了最大的reader数量。
好了，知道了RWMutex的设计方案和具体字段，下面我来解释一下具体的方法实现。
RLock/RUnlock的实现
首先，我们看一下移除了race等无关紧要的代码后的RLock和RUnlock方法：
func (rw *RWMutex) RLock() {
if atomic.AddInt32(&rw.readerCount, 1) < 0 {
// rw.readerCount是负值的时候，意味着此时有writer等待请求锁，因为writer优先级高，所以把后来的reader阻塞休眠
runtime_SemacquireMutex(&rw.readerSem, false, 0)
}
}
func (rw *RWMutex) RUnlock() {
if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
rw.rUnlockSlow(r) // 有等待的writer
}
}
func (rw *RWMutex) rUnlockSlow(r int32) {
if atomic.AddInt32(&rw.readerWait, -1) == 0 {
// 最后一个reader了，writer终于有机会获得锁了
runtime_Semrelease(&rw.writerSem, false, 1)
}
}
第2行是对reader计数加1。你可能比较困惑的是，readerCount怎么还可能为负数呢？其实，这是因为，readerCount这个字段有双重含义：
- 没有writer竞争或持有锁时，readerCount和我们正常理解的reader的计数是一样的；
- 但是，如果有writer竞争锁或者持有锁时，那么，readerCount不仅仅承担着reader的计数功能，还能够标识当前是否有writer竞争或持有锁，在这种情况下，请求锁的reader的处理进入第4行，阻塞等待锁的释放。
调用RUnlock的时候，我们需要将Reader的计数减去1（第8行），因为reader的数量减少了一个。但是，第8行的AddInt32的返回值还有另外一个含义。如果它是负值，就表示当前有writer竞争锁，在这种情况下，还会调用rUnlockSlow方法，检查是不是reader都释放读锁了，如果读锁都释放了，那么可以唤醒请求写锁的writer了。
当一个或者多个reader持有锁的时候，竞争锁的writer会等待这些reader释放完，才可能持有这把锁。打个比方，在房地产行业中有条规矩叫做“买卖不破租赁”，意思是说，就算房东把房子卖了，新业主也不能把当前的租户赶走，而是要等到租约结束后，才能接管房子。这和RWMutex的设计是一样的。当writer请求锁的时候，是无法改变既有的reader持有锁的现实的，也不会强制这些reader释放锁，它的优先权只是限定后来的reader不要和它抢。
所以，rUnlockSlow将持有锁的reader计数减少1的时候，会检查既有的reader是不是都已经释放了锁，如果都释放了锁，就会唤醒writer，让writer持有锁。
Lock
RWMutex是一个多writer多reader的读写锁，所以同时可能有多个writer和reader。那么，为了避免writer之间的竞争，RWMutex就会使用一个Mutex来保证writer的互斥。
一旦一个writer获得了内部的互斥锁，就会反转readerCount字段，把它从原来的正整数readerCount(>=0)修改为负数（readerCount-rwmutexMaxReaders），让这个字段保持两个含义（既保存了reader的数量，又表示当前有writer）。
我们来看下下面的代码。第5行，还会记录当前活跃的reader数量，所谓活跃的reader，就是指持有读锁还没有释放的那些reader。
func (rw *RWMutex) Lock() {
// 首先解决其他writer竞争问题
rw.w.Lock()
// 反转readerCount，告诉reader有writer竞争锁
r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
// 如果当前有reader持有锁，那么需要等待
if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
runtime_SemacquireMutex(&rw.writerSem, false, 0)
}
}
如果readerCount不是0，就说明当前有持有读锁的reader，RWMutex需要把这个当前readerCount赋值给readerWait字段保存下来（第7行）， 同时，这个writer进入阻塞等待状态（第8行）。
每当一个reader释放读锁的时候（调用RUnlock方法时），readerWait字段就减1，直到所有的活跃的reader都释放了读锁，才会唤醒这个writer。
Unlock
当一个writer释放锁的时候，它会再次反转readerCount字段。可以肯定的是，因为当前锁由writer持有，所以，readerCount字段是反转过的，并且减去了rwmutexMaxReaders这个常数，变成了负数。所以，这里的反转方法就是给它增加rwmutexMaxReaders这个常数值。
既然writer要释放锁了，那么就需要唤醒之后新来的reader，不必再阻塞它们了，让它们开开心心地继续执行就好了。
在RWMutex的Unlock返回之前，需要把内部的互斥锁释放。释放完毕后，其他的writer才可以继续竞争这把锁。
func (rw *RWMutex) Unlock() {
// 告诉reader没有活跃的writer了
r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
// 唤醒阻塞的reader们
for i := 0; i < int(r); i++ {
runtime_Semrelease(&rw.readerSem, false, 0)
}
// 释放内部的互斥锁
rw.w.Unlock()
}
在这段代码中，我删除了race的处理和异常情况的检查，总体看来还是比较简单的。这里有几个重点，我要再提醒你一下。首先，你要理解readerCount这个字段的含义以及反转方式。其次，你还要注意字段的更改和内部互斥锁的顺序关系。在Lock方法中，是先获取内部互斥锁，才会修改的其他字段；而在Unlock方法中，是先修改的其他字段，才会释放内部互斥锁，这样才能保证字段的修改也受到互斥锁的保护。
好了，到这里我们就完整学习了RWMutex的概念和实现原理。RWMutex的应用场景非常明确，就是解决readers-writers问题。学完了今天的内容，之后当你遇到这类问题时，要优先想到RWMutex。另外，Go并发原语代码实现的质量都很高，非常精炼和高效，所以，你可以通过看它们的实现原理，学习一些编程的技巧。当然，还有非常重要的一点就是要知道reader或者writer请求锁的时候，既有的reader/writer和后续请求锁的reader/writer之间的（释放锁/请求锁）顺序关系。
有个很有意思的事儿，就是官方的文档对RWMutex介绍是错误的，或者说是不明确的，在下一个版本（Go 1.16）中，官方会更改对RWMutex的介绍，具体是这样的：
A RWMutex is a reader/writer mutual exclusion lock.
The lock can be held by any number of readers or a single writer, and
a blocked writer also blocks new readers from acquiring the lock.
这个描述是相当精确的，它指出了RWMutex可以被谁持有，以及writer比后续的reader有获取锁的优先级。
虽然RWMutex暴露的API也很简单，使用起来也没有复杂的逻辑，但是和Mutex一样，在实际使用的时候，也会很容易踩到一些坑。接下来，我给你重点介绍3个常见的踩坑点。
RWMutex的3个踩坑点
坑点1：不可复制
前面刚刚说过，RWMutex是由一个互斥锁和四个辅助字段组成的。我们很容易想到，互斥锁是不可复制的，再加上四个有状态的字段，RWMutex就更加不能复制使用了。
不能复制的原因和互斥锁一样。一旦读写锁被使用，它的字段就会记录它当前的一些状态。这个时候你去复制这把锁，就会把它的状态也给复制过来。但是，原来的锁在释放的时候，并不会修改你复制出来的这个读写锁，这就会导致复制出来的读写锁的状态不对，可能永远无法释放锁。
那该怎么办呢？其实，解决方案也和互斥锁一样。你可以借助vet工具，在变量赋值、函数传参、函数返回值、遍历数据、struct初始化等时，检查是否有读写锁隐式复制的情景。
坑点2：重入导致死锁
读写锁因为重入（或递归调用）导致死锁的情况更多。
我先介绍第一种情况。因为读写锁内部基于互斥锁实现对writer的并发访问，而互斥锁本身是有重入问题的，所以，writer重入调用Lock的时候，就会出现死锁的现象，这个问题，我们在学习互斥锁的时候已经了解过了。
func foo(l *sync.RWMutex) {
fmt.Println("in foo")
l.Lock()
bar(l)
l.Unlock()
}
func bar(l *sync.RWMutex) {
l.Lock()
fmt.Println("in bar")
l.Unlock()
}
func main() {
l := &sync.RWMutex{}
foo(l)
}
运行这个程序，你就会得到死锁的错误输出，在Go运行的时候，很容易就能检测出来。
第二种死锁的场景有点隐蔽。我们知道，有活跃reader的时候，writer会等待，如果我们在reader的读操作时调用writer的写操作（它会调用Lock方法），那么，这个reader和writer就会形成互相依赖的死锁状态。Reader想等待writer完成后再释放锁，而writer需要这个reader释放锁之后，才能不阻塞地继续执行。这是一个读写锁常见的死锁场景。
第三种死锁的场景更加隐蔽。
当一个writer请求锁的时候，如果已经有一些活跃的reader，它会等待这些活跃的reader完成，才有可能获取到锁，但是，如果之后活跃的reader再依赖新的reader的话，这些新的reader就会等待writer释放锁之后才能继续执行，这就形成了一个环形依赖： writer依赖活跃的reader -> 活跃的reader依赖新来的reader -> 新来的reader依赖writer。
这个死锁相当隐蔽，原因在于它和RWMutex的设计和实现有关。啥意思呢？我们来看一个计算阶乘(n!)的例子：
func main() {
var mu sync.RWMutex
// writer,稍微等待，然后制造一个调用Lock的场景
go func() {
time.Sleep(200 * time.Millisecond)
mu.Lock()
fmt.Println("Lock")
time.Sleep(100 * time.Millisecond)
mu.Unlock()
fmt.Println("Unlock")
}()
go func() {
factorial(&mu, 10) // 计算10的阶乘, 10!
}()
select {}
}
// 递归调用计算阶乘
func factorial(m *sync.RWMutex, n int) 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 // 递归调用
}
factoria方法是一个递归计算阶乘的方法，我们用它来模拟reader。为了更容易地制造出死锁场景，我在这里加上了sleep的调用，延缓逻辑的执行。这个方法会调用读锁（第27行），在第33行递归地调用此方法，每次调用都会产生一次读锁的调用，所以可以不断地产生读锁的调用，而且必须等到新请求的读锁释放，这个读锁才能释放。
同时，我们使用另一个goroutine去调用Lock方法，来实现writer，这个writer会等待200毫秒后才会调用Lock，这样在调用Lock的时候，factoria方法还在执行中不断调用RLock。
这两个goroutine互相持有锁并等待，谁也不会退让一步，满足了“writer依赖活跃的reader -> 活跃的reader依赖新来的reader -> 新来的reader依赖writer”的死锁条件，所以就导致了死锁的产生。
所以，使用读写锁最需要注意的一点就是尽量避免重入，重入带来的死锁非常隐蔽，而且难以诊断。
坑点3：释放未加锁的RWMutex
和互斥锁一样，Lock和Unlock的调用总是成对出现的，RLock和RUnlock的调用也必须成对出现。Lock和RLock多余的调用会导致锁没有被释放，可能会出现死锁，而Unlock和RUnlock多余的调用会导致panic。在生产环境中出现panic是大忌，你总不希望半夜爬起来处理生产环境程序崩溃的问题吧？所以，在使用读写锁的时候，一定要注意，不遗漏不多余。
流行的Go开发项目中的坑
好了，又到了泡一杯宁夏枸杞加新疆大滩枣的养生茶，静静地欣赏知名项目出现Bug的时候了，这次被拉出来的是RWMutex的Bug。
Docker
issue 36840
issue 36840修复的是错误地把writer当成reader的Bug。 这个地方本来需要修改数据，需要调用的是写锁，结果用的却是读锁。或许是被它紧挨着的findNode方法调用迷惑了，认为这只是一个读操作。可实际上，代码后面还会有changeNodeState方法的调用，这是一个写操作。修复办法也很简单，只需要改成Lock/Unlock即可。
Kubernetes
issue 62464
issue 62464就是读写锁第二种死锁的场景，这是一个典型的reader导致的死锁的例子。知道墨菲定律吧？“凡是可能出错的事，必定会出错”。你可能觉得我前面讲的RWMutex的坑绝对不会被人踩的，因为道理大家都懂，但是你看，Kubernetes就踩了这个重入的坑。
这个issue在移除pod的时候可能会发生，原因就在于，GetCPUSetOrDefault方法会请求读锁，同时，它还会调用GetCPUSet或GetDefaultCPUSet方法。当这两个方法都请求写锁时，是获取不到的，因为GetCPUSetOrDefault方法还没有执行完，不会释放读锁，这就形成了死锁。
总结
在开发过程中，一开始考虑共享资源并发访问问题的时候，我们就会想到互斥锁Mutex。因为刚开始的时候，我们还并不太了解并发的情况，所以，就会使用最简单的同步原语来解决问题。等到系统成熟，真正到了需要性能优化的时候，我们就能静下心来分析并发场景的可能性，这个时候，我们就要考虑将Mutex修改为RWMutex，来压榨系统的性能。
当然，如果一开始你的场景就非常明确了，比如我就要实现一个线程安全的map，那么，一开始你就可以考虑使用读写锁。
正如我在前面提到的，如果你能意识到你要解决的问题是一个readers-writers问题，那么你就可以毫不犹豫地选择RWMutex，不用考虑其它选择。那在使用RWMutex时，最需要注意的一点就是尽量避免重入，重入带来的死锁非常隐蔽，而且难以诊断。
另外我们也可以扩展RWMutex，不过实现方法和互斥锁Mutex差不多，在技术上是一样的，都是通过unsafe来实现，我就不再具体讲了。课下你可以参照我们上节课学习的方法，实现一个扩展的RWMutex。
这一讲我们系统学习了读写锁的相关知识，这里提供给你一个知识地图，帮助你复习本节课的知识。
思考题
请你写一个扩展的读写锁，比如提供TryLock，查询当前是否有writer、reader的数量等方法。
欢迎在留言区写下你的思考和答案，我们一起交流讨论。如果你觉得有所收获，也欢迎你把今天的内容分享给你的朋友或同事。