Go语言读写锁RWMutex的源码分析

在前面两篇文章中初见 Go Mutex、Go Mutex 源码详解，我们学习了Go语言中的Mutex，它是一把互斥锁，每次只允许一个goroutine进入临界区，可以保证临界区资源的状态正确性。但是有的情况下，并不是所有goroutine都会修改临界区状态，可能只是读取临界区的数据，如果此时还是需要每个goroutine拿到锁依次进入的话，效率就有些低下了。例如房间里面有一幅画，有人想修改，有人只是想看一下，完全可以放要看的一部分人进去，等他们看完再让修改的人进去修改，这样既提高了效率，也保证了临界区资源的安全。看和修改，对应的就是读和写，本篇文章我们就一起来学习下Go语言中的读写锁sync.RWMutex。

说明：本文中的示例，均是基于Go1.17 64位机器

RWMutex 总览

RWMutex是一个读/写互斥锁，在某一时刻只能由任意数量的reader持有或者一个 writer持有。也就是说，要么放行任意数量的 reader，多个 reader 可以并行读；要么放行一个 writer，多个 writer 需要串行写。

RWMutex 对外暴露的方法有五个：

RLock()：读操作获取锁，如果锁已经被writer占用，会一直阻塞直到writer释放锁；否则直接获得锁；
RUnlock()：读操作完毕之后释放锁；
Lock()：写操作获取锁，如果锁已经被reader或者writer占用，会一直阻塞直到获取到锁；否则直接获得锁；
Unlock()：写操作完毕之后释放锁；
RLocker()：返回读操作的Locker对象，该对象的Lock()方法对应RWMutex的RLock()，Unlock()方法对应RWMutex的RUnlock()方法。

一旦涉及到多个 reader 和 writer ，就需要考虑优先级问题，是 reader 优先还是 writer 优先：

reader优先：只要有 reader 要进行读操作，writer 就一直等待，直到没有 reader 到来。这种方式做到了读操作的并发，但是如果 reader 持续到来，会导致 writer 饥饿，一直不能进行写操作；
writer优先：只要有 writer 要进行写操作，reader 就一直等待，直到没有 writer 到来。这种方式提高了写操作的优先级，但是如果 writer 持续到来，会导致 reader 饥饿，一直不能进行读操作；
没有优先级：按照先来先到的顺序，没有谁比谁更优先，这种相对来说会更公平。

我们先来看下 RWMutex 的运行机制，就可以知道它的优先级是什么了。

可以想象 RWMutex 有两个队伍，一个是包含所有reader和你获得准入权writer的队列A，一个是还没有获得准入权 writer 的队列B。

队列 A 最多只允许有一个writer，如果有其他 writer，需要在队列B 等待；
当一个 writer 到了队列A 后，只允许它之前的reader执行读操作，新来的 reader 需要在队列A 后面排队；
当前面的 reader 执行完读操作之后，writer 执行写操作；
writer 执行完写操作后，让后面的reader执行读操作，再唤醒队列B的一个writer到队列A 后面排队。

初始时刻队列A 中 writerW1前面有三个 reader，后面有两个 reader，队列B中有两个 writer

RWMutex运行示例：初始时刻

并发读多个 reader 可以同时获取到读锁，进入临界区进行读操作；writerW1在队列A中等待，同时又来了两个 reader，直接在队列A后面排队

RWMutex运行示例：并发读

写操作W1前面所有的 reader 完成后，W1获得锁，进入临界区操作

RWMutex运行示例：写操作

获得准入权W1完成写操作退出，先让后面排队的 reader 进行读操作，然后从队列B 中唤醒W2到队列A排队。W2从队列B到队列A的过程中，R8先到了队列A，因此R8可以执行读操作。R9、R10、R11在W2之后到的，所以在后面排队；新来的W4直接在队列B 排队。

RWMutex运行示例：获得准入权

从上面的示例可以看出，RWMutex可以看作是没有优先级，按照先来先到的顺序去执行，只不过是多个reader可以并行去执行罢了。

深入源码

数据结构

type RWMutex struct {
 w           Mutex  // 控制 writer 在 队列B 排队
 writerSem   uint32 // 写信号量，用于等待前面的 reader 完成读操作
 readerSem   uint32 // 读信号量，用于等待前面的 writer 完成写操作
 readerCount int32  // reader 的总数量，同时也指示是否有 writer 在队列A 中等待
 readerWait  int32  // 队列A 中 writer 前面 reader 的数量
}

// 允许最大的 reader 数量
const rwmutexMaxReaders = 1 << 30

上述中的几个变量，比较特殊的是readerCount，不仅表示当前所有reader的数量，同时表示是否有writer在队列A中等待。当readerCount变为负数时，就代表有 writer 在队列A 中等待了。

当有 writer 进入队列A 后，会将readerCount变为负数，即readerCount = readerCount - rwmutexMaxReaders，同时利用readerWait变量记录它前面有多少个 reader；
如果有新来的 reader，发现readerCount是负数，就会直接去后面排队；
writer 前面的 reader 在释放锁时，会将readerCount和readerWait都减一，当 readerWait==0 时，表示 writer 前面的所有 reader 都执行完了，可以让 writer 执行写操作了；
writer 执行写操作完毕后，会将 readerCount 再变回正数，readerCount = readerCount + rwmutexMaxReaders。

举例：假设当前有两个 reader，readerCount = 2；允许最大的reader 数量为 10

当 writer 进入队列A 时，readerCount = readerCount - rwmutexMaxReaders = -8，readerWait = readerCount = 2
如果再来 3 个reader，readerCount = readerCount + 3 = -5
获得读锁的两个reader 执行完后，readerCount = readerCount - 2 = -7，readerWait = readerWait-2 =0，writer 获得锁
writer 执行完后，readerCount = readerCount + rwmutexMaxReaders = 3，当前有 3个 reader

RLock()

reader 执行读操作之前，需要调用 RLock() 获取锁

func (rw *RWMutex) RLock() {
 
  // reader 加锁，将 readerCount 加一，表示多了个 reader
  if atomic.AddInt32(&rw.readerCount, 1) < 0 {
  
    // 如果 readerCount<0，说明有 writer 在自己前面等待，排队等待读信号量
  runtime_SemacquireMutex(&rw.readerSem, false, 0)
 }
}

RUnlock()

reader 执行完读操作后，调用 RUnlock() 释放锁

func (rw *RWMutex) RUnlock() {

  // reader 释放锁，将 readerCount 减一，表示少了个 reader
 if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
  
    // 如果readerCount<0，说明有 writer 在自己后面等待，看是否要让 writer 运行
  rw.rUnlockSlow(r)
 }
}

func (rw *RWMutex) rUnlockSlow(r int32) {

  // 将 readerWait 减一，表示前面的 reader 少了一个
 if atomic.AddInt32(&rw.readerWait, -1) == 0 {
    
  // 如果 readerWait 变为了0，那么自己就是最后一个完成的 reader
    // 释放写信号量，让 writer 运行
  runtime_Semrelease(&rw.writerSem, false, 1)
 }
}

Lock()

writer 执行写操作之前，调用 Lock() 获取锁

func (rw *RWMutex) Lock() {

   // 利用互斥锁，如果前面有 writer，那么就需要等待互斥锁，即在队列B 中排队等待；如果没有，可以直接进入 队列A 排队
   rw.w.Lock()

  // atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) 将 readerCount 变成了负数
  // 再加 rwmutexMaxReaders，相当于 r = readerCount，r 就是 writer 前面的 reader 数量
   r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
   
   // 如果 r!= 0 ，表示自己前面有 reader，那么令 readerWait = r，要等前面的 reader 运行完
   if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
      runtime_SemacquireMutex(&rw.writerSem, false, 0)
   }

}

Lock()和RUnlock()是会并发进行的：

如果 Lock() 将 readerCount 变为负数后，假设 r=3，表示加入的那一刻前面有三个 reader，还没有赋值 readerWait CPU 就被强占了，readerWait = 0；
假设此时三个 reader 的 RUnlock() 会进入到 rUnlockSlow() 逻辑，每个 reader 都将 readerWait 减一， readerWait 会变成负数，此时不符合唤醒 writer 的条件；
三个 reader 运行完之后，此时 readerWait = -3， Lock() 运行到 atomic.AddInt32(&rw.readerWait, r) = -3+3 =0，也不会休眠，直接获取到锁，因为前面的 reader 都运行完了。

这就是为什么rUnlockSlow()要判断atomic.AddInt32(&rw.readerWait, -1) == 0以及Lock()要判断atomic.AddInt32(&rw.readerWait, r) != 0的原因。

Unlock()

writer 执行写操作之后，调用 Lock() 释放锁

func (rw *RWMutex) Unlock() {

  // 将 readerCount 变为正数，表示当前没有 writer 在队列A 等待了
 r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)

  // 将自己后面等待的 reader 唤醒，可以进行读操作了
 for i := 0; i < int(r); i++ {
  runtime_Semrelease(&rw.readerSem, false, 0)
 }
 
  // 释放互斥锁，如果队列B有writer，相当于唤醒一个来队列A 排队
 rw.w.Unlock()

}

writer 对 readerCount 一加一减，不会改变整体状态，只是用正负来表示是否有 writer 在等待。当然，如果在 writer 将 readerCount变为负数后，来了很多 reader，将 readerCount 变为了正数，此时reader 在 writer 没有释放锁的时候就获取到锁了，是有问题的。但是 rwmutexMaxReaders 非常大，可以不考虑这个问题。

常见问题

不可复制

和 Mutex 一样，RWMutex 也是不可复制。不能复制的原因和互斥锁一样。一旦读写锁被使用，它的字段就会记录它当前的一些状态。这个时候你去复制这把锁，就会把它的状态也给复制过来。但是，原来的锁在释放的时候，并不会修改你复制出来的这个读写锁，这就会导致复制出来的读写锁的状态不对，可能永远无法释放锁。

不可重入

不可重入的原因是，获得锁之后，还没释放锁，又申请锁，这样有可能造成死锁。比如 reader A 获取到了读锁，writer B 等待 reader A 释放锁，reader 还没释放锁又申请了一把锁，但是这把锁申请不成功，他需要等待 writer B。这就形成了一个循环等待的死锁。

加锁和释放锁一定要成对出现，不能忘记释放锁，也不能解锁一个未加锁的锁。

实战一下

Go 中的 map 是不支持并发写的，我们可以利用读写锁 RWMutex 来实现并发安全的 map。在读多写少的情况下，使用 RWMutex 要比 Mutex 性能高。

package main

import (
 "fmt"
 "math/rand"
 "sync"
 "time"
)

type ConcurrentMap struct {
 m     sync.RWMutex
 items map[string]interface{}
}

func (c *ConcurrentMap) Add(key string, value interface{}) {
 c.m.Lock()
 defer c.m.Unlock()
 c.items[key] = value
}

func (c *ConcurrentMap) Remove(key string) {
 c.m.Lock()
 defer c.m.Unlock()
 delete(c.items, key)
}
func (c *ConcurrentMap) Get(key string) interface{} {
 c.m.RLock()
 defer c.m.RUnlock()
 return c.items[key]
}

func NewConcurrentMap() ConcurrentMap {
 return ConcurrentMap{
  items: make(map[string]interface{}),
 }
}

func main() {
 m := NewConcurrentMap()

 var wait sync.WaitGroup

 wait.Add(10000)
 for i := 0; i < 10000; i++ {

  key := fmt.Sprintf("%d", rand.Intn(10))
  value := fmt.Sprintf("%d", rand.Intn(100))
  if i%100 == 0 {
   go func() {
    defer wait.Done()
    m.Add(key, value)
   }()
  } else {
   go func() {
    defer wait.Done()
    fmt.Println(m.Get(key))
    time.Sleep(time.Millisecond * 10)
   }()
  }

 }

 wait.Wait()
}