并发编程(五) - RWMutex: 读写锁

问题引入

有一个全局的配置，线上每次请求「qps 1w」都会使用到这个配置，且每隔 1min 会更新一次配置。为了保证并发场景下数据的正常读取，可以选择使用 Mutex 对读写数据进行保护。

type config struct {
    name string
    m    sync.Mutex
}

func (c *config) update(name string) {
    c.m.Lock()
    defer c.m.Unlock()
    c.name = name
}

func (c *config) read() string {
    c.m.Lock()
    defer c.m.Unlock()
    return c.name
}

func main() {
    var c config

    go func() { //定时更新
t := time.Tick(time.Second)
       for {
          select {
          case <-t:
             c.update(time.Now().String())
          }
       }
    }()

    for i := 0; i < 10000; i++ {
       go func() {
          c.read()
       }()
    }

    time.Sleep(time.Minute)

}

在上面的例子中，由于使用了 Mutex，会保证每次读或者写都只能由一个 goroutine 执行。然而，对于读操作，并发读取并不会导致数据异常。可以考虑允许读操作并发执行，读写操作互斥，写写操作互斥，从而提升服务性能。

使用示例

针对上面的问题，提出了读写锁来进一步提升服务的吞吐能力。假设有一个 goroutine 获取了读锁进行读操作；又来了一个 goroutine 想执行读操作，无需等待获取读锁的 goroutine 释放读锁，可以直接获取到读锁进行读操作；当有goroutine 获取到写锁时，其他 goroutine 获取读锁/写锁时均被阻塞。读写锁和互斥锁的本质区别就在于多个 goroutine 可以同时获取读锁，而对于写锁，只能有一个 goroutine 获取写锁，且写锁和读锁互斥。

在 go 的基础库中实现了一个 RWMutex。提供了以下方法，方法声明比较清楚，方便开发者直接使用。

Type RWMutex
// 读操作调用，获取读锁，获取不到则阻塞
func (rw *RWMutex) RLock()
// 尝试获取读锁，不阻塞
func (rw *RWMutex) TryRLock() bool
// 释放读锁
func (rw *RWMutex) RUnlock()
// 获取写锁，获取不到则阻塞
func (rw *RWMutex) Lock()
// 尝试获取写锁，不阻塞
func (rw *RWMutex) TryLock()
// 释放写锁
func (rw *RWMutex) Unlock()
// 将读锁转化为 Locker 接口，底层调用读锁的 RLock 和 UnLock 方法
func (rw *RWMutex) RLocker() Locker

使用读写锁对上面的例子进行改造，改造也比较简单。数据读取的时候使用读锁，数据变更的时候使用写锁。

type config struct {
    name string
    rw   sync.RWMutex
}

func (c *config) update(name string) {
    c.rw.Lock()
    defer c.rw.Unlock()
    c.name = name
}

func (c *config) read() string {
    c.rw.RLock()
    defer c.rw.RUnlock()
    return c.name
}

实现原理

在实现读写锁之前需要先考虑读写锁获取优先级问题。举个具体的例子，当有 goroutine 已经获取到了读锁，这是有两个 goroutine 一个要获取读锁，一个要获取写锁，那谁应该被阻塞呢？针对优先级的不同，可以将读写锁分为三类：

读优先：即优先获取读锁，写锁被阻塞。优点是提升了读接口的并发性能。缺点是容易造成写饥饿，即持续获取不到写锁，导致数据无法更新。

写优先。 即优先获取写锁，后续的读锁均进入阻塞状态。优点是避免写饥饿的问题，可以保证数据及时的更新。

不指定优先级。 即不指定明确的优先级，按照队列的形式获取锁。根据当时的调度，如果先来的是读锁则直接获取读锁，如果先来的是写锁，则阻塞等待上一个 goroutine 获取写锁。这种场景下，实现起来相对比较复杂。

在 go 中采取的是写优先的机制。即如果已经有 goroutine 获取了读锁，申请写锁后会阻塞后续的读锁，直到写锁释放。

先看一下 RWMutex 的结构体定义，使用 Mutex 用于实现 writer 之间的互斥逻辑；使用信号量实现读写锁 goroutine 的阻塞等待；readerCount 存储 reader 的数量，这里需要注意的是当有 writer 进来时会将 readerCount 翻转为负数「readerCount-rwmutexMaxReaders」，用于表示有 writer 阻塞，从而阻塞后续的 reader；readerWait 用于表示 writer 需要等待释放 reader 的数量，当值为 0 时唤醒阻塞的 writer。

type RWMutex struct {
    w           Mutex        // held if there are pending writers
    writerSem   uint32       // semaphore for writers to wait for completing readers
    readerSem   uint32       // semaphore for readers to wait for completing writers
    readerCount atomic.Int32 // number of pending readers
    readerWait  atomic.Int32 // number of departing readers
}

获取读锁。先获取一个互斥锁，保证 writer 与 writer 之间的互斥。将 readerCount 翻转为负数，标识有 writer 进来，阻塞后续的 reader。判断当 reader 的数量，设置 readerWait，标识 writer 需要等待读锁释放的数量。这里有一个巧妙的设计，就是使用 readerCount 的翻转表示两种含义，正数表示没有 writer 进来，其值为 reader 的数量；负数表示有 writer 进来，阻塞后续的 reader，同时利用 rw.readerCount.Add(-rwmutexMaxReaders) + rwmutexMaxReaders 计算 reader 的数量。

func (rw *RWMutex) Lock() {
    // First, resolve competition with other writers.
    rw.w.Lock()
    // Announce to readers there is a pending writer.
    r := rw.readerCount.Add(-rwmutexMaxReaders) + rwmutexMaxReaders
    // Wait for active readers.
    if r != 0 && rw.readerWait.Add(r) != 0 {
         runtime_SemacquireRWMutex(&rw.writerSem, false, 0)
    }
}

释放写锁。翻转 readerCount，将其置为正数，其他 reader 可以开始获取读锁。释放 readerSem，唤醒阻塞的 reader，释放 mutex 允许其他的 writer 开始处理。

func (rw *RWMutex) Unlock() {
    // Announce to readers there is no active writer.
    r := rw.readerCount.Add(rwmutexMaxReaders)
    // 多次释放写锁
    if r >= rwmutexMaxReaders {
       race.Enable()
       fatal("sync: Unlock of unlocked RWMutex")
    }
    // Unblock blocked readers, if any.
    for i := 0; i < int(r); i++ {
       runtime_Semrelease(&rw.readerSem, false, 0)
    }
    // Allow other writers to proceed.
    rw.w.Unlock()
}

获取读锁。更新 reader 的数量。当 readerCount 为负值时，表示有 writer 执行，使用 readerSem 信号量阻塞，等待 writer 释放。

func (rw *RWMutex) RLock() {
    if rw.readerCount.Add(1) < 0 {
       // A writer is pending, wait for it.
        runtime_SemacquireRWMutexR(&rw.readerSem, false, 0)
    }
}

释放读锁。更新 reader 数量。当 readerCount 为负值时，表示有 writer 执行，进入 rUnlockSlow。在 rUnlockSlow 中，先判断是否重复调用了 RUnlock。由于有 writer 等待，更新需要等待读锁释放的个数 readerWait，当值为 0 时唤醒写锁。

func (rw *RWMutex) RUnlock() {
    if r := rw.readerCount.Add(-1); r < 0 {
       // Outlined slow-path to allow the fast-path to be inlined
    rw.rUnlockSlow(r)
    }
}

func (rw *RWMutex) rUnlockSlow(r int32) {
    if r+1 == 0 || r+1 == -rwmutexMaxReaders {
       fatal("sync: RUnlock of unlocked RWMutex")
    }
    // A writer is pending.
    if rw.readerWait.Add(-1) == 0 {
       // The last reader unblocks the writer.
        runtime_Semrelease(&rw.writerSem, false, 1)
    }
}

注意事项

其实所有的注意事项都在源码中有所体现，当有非预期的逻辑时会直接 fatal。

不可复制

和 Mutex 一样，RWMutex 的实现依赖内部的状态，因此多个 goroutine 在使用时不能使用复制的值。由于 RWMutex 也实现了 Lock 和 Ulock 接口，可以直接被 vet 工具检测到程序中的复制行为。

不可重入

RWMutex 是不可重入的，因此在同一个 goroutine 重复调用 Lock 或 write 中调用 reader 与 reader 中调用 writer 均会导致死锁。

// 写锁不可重入
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)
}

// 读锁获取写锁，写锁又获取读锁，造成死锁
func main() {
    l := &sync.RWMutex{}
    l.RLock()
    defer l.Unlock()
    l.Lock()
    defer l.Unlock()

}

// 写锁获取读锁，造成死锁
func main() {
    l := &sync.RWMutex{}
    l.Lock()
    defer l.Lock()
    l.RLock()
    defer l.RUnlock()
}

释放未加锁的 RWMutex

和互斥锁一样，Lock 和 Unlock 的调用总是成对出现的，RLock 和 RUnlock 的调用也必须成对出现。Lock 和 RLock 多余的调用会导致锁没有被释放，可能会出现死锁，而 Unlock 和 RUnlock 多余的调用会导致 panic。

总结

对于互斥的场景，使用 Mutex 肯定没有任何问题，它保证了对应的操作在任何时刻只能被一个 goroutine 所执行。然而对于读多写少的场景，每次操作都互斥显然成本是有点高的。假设程序中有 A、B 两个操作，A 和 B 两个操作是互斥的，A 操作之间是可以并行的，B 操作之间是互斥的。满足这种场景的情况下可以使用 RWMutex，但要不要使用 RWMutex 还需要考虑 A 并发执行的次数是否远大于 B 执行的次数。如果 A B 操作并发执行的次数接近，使用 Mutex 性能反而更好。