深入理解 go RWMutex

在上一篇文章《深入理解 go Mutex》中，我们已经对 go Mutex 的实现原理有了一个大致的了解，也知道了 Mutex 可以实现并发读写的安全。今天，我们再来看看另外一种锁，RWMutex，有时候，其实我们读数据的频率要远远高于写数据的频率，而且不同协程应该可以同时读取的，这个时候，RWMutex 就派上用场了。

RWMutex 的实现原理和 Mutex 类似，只是在 Mutex 的基础上，区分了读锁和写锁：

读锁：只要没有写锁，就可以获取读锁，多个协程可以同时获取读锁（可以并行读）。
写锁：只能有一个协程获取写锁，其他协程想获取读锁或写锁都只能等待。

下面就让我们来深入了解一下 RWMutex 的基本使用和实现原理等内容。

RWMutex 的整体模型

正如 RWMutex 的命名那样，它是区分了读锁和写锁的锁，所以我们可以从读和写两个方面来看 RWMutex 的模型。

下文中的 reader 指的是进行读操作的 goroutine，writer 指的是进行写操作的 goroutine。

读操作模型

我们可以用下图来表示 RWMutex 的读操作模型：

上图使用了 w.Lock，是因为 RWMutex 的实现中，写锁是使用 Mutex 来实现的。

说明：

读操作的时候可以同时有多个 goroutine 持有 RLock，然后进入临界区。（也就是可以并行读），上图的 G1、G2 和 G3 就是同时持有 RLock 的几个 goroutine。
在读操作的时候，如果有 goroutine 持有 RLock，那么其他 goroutine （不管是读还是写）就只能等待，直到所有持有 RLock 的 goroutine 释放锁。
也就是上图的 G4 需要等待 G1、G2 和 G3 释放锁之后才能进入临界区。
最后，因为 G5 和 G6 这两个协程获取锁的时机比 G4 晚，所以它们会在 G4 释放锁之后才能进入临界区。

写操作模型

我们可以用下图来表示 RWMutex 的写操作模型：

说明：

写操作的时候只能有一个 goroutine 持有 Lock，然后进入临界区，释放写锁之前，所有其他的 goroutine 都只能等待。
上图的 G1~G5 表示的是按时间顺序先后获取锁的几个 goroutine。
上面几个 goroutine 获取锁的过程是：
- G1 获取写锁，进入临界区。然后 G2、G3、G4 和 G5 都在等待。
- G1 释放写锁之后，G2 和 G3 可以同时获取读锁，进入临界区。然后 G3、G4 和 G5 都在等待。
- G2 和 G3 可以同时获取读锁，进入临界区。然后 G4 和 G5 都在等待。
- G2 和 G3 释放读锁之后，G4 获取写锁，进入临界区。然后 G5 在等待。
- 最后，G4 释放写锁，G5 获取读锁，进入临界区。

基本用法

RWMutex 中包含了以下的方法：

Lock：获取写锁，如果有其他 goroutine 持有读锁或写锁，那么就会阻塞等待。
Unlock：释放写锁。
RLock：获取读锁，如果有其他 goroutine 持有写锁，那么就会阻塞等待。
RUnlock：释放读锁。

其他不常用的方法：

RLocker：返回一个读锁，该锁包含了 RLock 和 RUnlock 方法，可以用来获取读锁和释放读锁。
TryLock: 尝试获取写锁，如果获取成功，返回 true，否则返回 false。不会阻塞等待。
TryRLock: 尝试获取读锁，如果获取成功，返回 true，否则返回 false。不会阻塞等待。

一个简单的例子

我们可以通过下面的例子来看一下 RWMutex 的基本用法：

package mutex

import (
	"sync"
	"testing"
)

var config map[string]string
var mu sync.RWMutex

func TestRWMutex(t *testing.T) {
	config = make(map[string]string)

	// 启动 10 个 goroutine 来写
	var wg1 sync.WaitGroup
	wg1.Add(10)
	for i := 0; i < 10; i++ {
		go func() {
			set("foo", "bar")
			wg1.Done()
		}()
	}

	// 启动 100 个 goroutine 来读
	var wg2 sync.WaitGroup
	wg2.Add(100)
	for i := 0; i < 100; i++ {
		go func() {
			get("foo")
			wg2.Done()
		}()
	}

	wg1.Wait()
	wg2.Wait()
}

// 获取配置
func get(key string) string {
	// 获取读锁，可以多个 goroutine 并发读取
	mu.RLock()
	defer mu.RUnlock()

	if v, ok := config[key]; ok {
		return v
	}

	return ""
}

// 设置配置
func set(key, val string) {
	// 获取写锁
	mu.Lock()
	defer mu.Unlock()

	config[key] = val
}

上面的例子中，我们启动了 10 个 goroutine 来写配置，启动了 100 个 goroutine 来读配置。这跟我们现实开发中的场景是一样的，很多时候其实是读多写少的。如果我们在读的时候也使用互斥锁，那么就会导致读的性能非常差，因为读操作一般都不会有副作用的，但是如果使用互斥锁，那么就只能一个一个的读了。

而如果我们使用 RWMutex，那么就可以同时有多个 goroutine 来读取配置，这样就可以大大提高读的性能。因为我们进行读操作的时候，可以多个 goroutine 并发读取，这样就可以大大提高读的性能。

RWMutex 使用的注意事项

在《深入理解 go Mutex》中，我们已经讲过了 Mutex 的使用注意事项，其实 RWMutex 的使用注意事项也是差不多的：

不要忘记释放锁，不管是读锁还是写锁。
Lock 之后，没有释放锁之前，不能再次使用 Lock。
在 Unlock 之前，必须已经调用了 Lock，否则会 panic
在第一次使用 RWMutex 之后，不能复制，因为这样一来 RWMutex 的状态也会被复制。这个可以使用 go vet 来检查。

源码剖析

RWMutex 的一些实现原理跟 Mutex 是一样的，比如阻塞的时候使用信号量等，在 Mutex 那一篇中已经有讲解了，这里不再赘述。这里就 RWMutex 的实现原理进行一些简单的剖析。

RWMutex 结构体

RWMutex 的结构体定义如下：

type RWMutex struct {
	w           Mutex        // 互斥锁，用于保护读写锁的状态
	writerSem   uint32       // writer 信号量
	readerSem   uint32       // reader 信号量
	readerCount atomic.Int32 // 所有 reader 数量
	readerWait  atomic.Int32 // writer 等待完成的 reader 数量
}

各字段含义：

w：互斥锁，用于保护读写锁的状态。RWMutex 的写锁是互斥锁，所以直接使用 Mutex 就可以了。
writerSem：writer 信号量，用于实现写锁的阻塞等待。
readerSem：reader 信号量，用于实现读锁的阻塞等待。
readerCount：所有 reader 数量（包括已经获取读锁的和正在等待获取读锁的 reader）。
readerWait：writer 等待完成的 reader 数量（也就是获取写锁的时刻，已经获取到读锁的 reader 数量）。

因为要区分读锁和写锁，所以在 RWMutex 中，我们需要两个信号量，一个用于实现写锁的阻塞等待，一个用于实现读锁的阻塞等待。我们需要特别注意的是 readerCount 和 readerWait 这两个字段，我们可能会比较好奇，为什么有了 readerCount 这个字段，还需要 readerWait 这个字段呢？

这是因为，我们在尝试获取写锁的时候，可能会有多个 reader 正在使用读锁，这时候我们需要知道有多少个 reader 正在使用读锁，等待这些 reader 释放读锁之后，就获取写锁了，而 readerWait 这个字段就是用来记录这个数量的。在 Lock 中获取写锁的时候，如果观测到 readerWait 不为 0 则会阻塞等待，直到 readerWait 为 0 之后才会真正获取写锁，然后才可以进行写操作。

读锁源码剖析

获取读锁的方法如下：

// 获取读锁
func (rw *RWMutex) RLock() {
	if rw.readerCount.Add(1) < 0 {
		// 有 writer 在使用锁，阻塞等待 writer 完成
		runtime_SemacquireRWMutexR(&rw.readerSem, false, 0)
	}
}

读锁的实现很简单，先将 readerCount 加 1，如果加 1 之后的值小于 0，说明有 writer 正在使用锁，那么就需要阻塞等待 writer 完成。

释放读锁的方法如下：

// 释放读锁
func (rw *RWMutex) RUnlock() {
	// readerCount 减 1，如果 readerCount 小于 0 说明有 writer 在等待
	if r := rw.readerCount.Add(-1); r < 0 {
		// 有 writer 在等待，唤醒 writer
		rw.rUnlockSlow(r)
	}
}

// 唤醒 writer
func (rw *RWMutex) rUnlockSlow(r int32) {
	// 未 Lock 就 Unlock，panic
	if r+1 == 0 || r+1 == -rwmutexMaxReaders {
		fatal("sync: RUnlock of unlocked RWMutex")
	}
	// readerWait 减 1，返回值是新的 readerWait 值
	if rw.readerWait.Add(-1) == 0 {
		// 最后一个 reader 唤醒 writer
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

读锁的实现总结：

获取读锁的时候，会将 readerCount 加 1
如果正在获取读锁的时候，发现 readerCount 小于 0，说明有 writer 正在使用锁，那么就需要阻塞等待 writer 完成。
释放读锁的时候，会将 readerCount 减 1
如果 readerCount 减 1 之后小于 0，说明有 writer 正在等待，那么就需要唤醒 writer。
唤醒 writer 的时候，会将 readerWait 减 1，如果 readerWait 减 1 之后为 0，说明 writer 获取锁的时候存在的 reader 都已经释放了读锁，可以获取写锁了。

·rwmutexMaxReaders算是一个特殊的标识，在获取写锁的时候会将readerCount的值减去rwmutexMaxReaders，所以在其他地方可以根据readerCount` 是否小于 0 来判断是否有 writer 正在使用锁。

写锁源码剖析

获取写锁的方法如下：

// 获取写锁
func (rw *RWMutex) Lock() {
	// 首先，解决与其他写入者的竞争。
	rw.w.Lock()
	// 向读者宣布有一个待处理的写入。
	// r 就是当前还没有完成的读操作，等这部分读操作完成之后才可以获取写锁。
	r := rw.readerCount.Add(-rwmutexMaxReaders) + rwmutexMaxReaders
	// 等待活跃的 reader
	if r != 0 && rw.readerWait.Add(r) != 0 {
		// 阻塞，等待最后一个 reader 唤醒
		runtime_SemacquireRWMutex(&rw.writerSem, false, 0)
	}
}

释放写锁的方法如下：

// 释放写锁
func (rw *RWMutex) Unlock() {
	// 向 readers 宣布没有活动的 writer。
	r := rw.readerCount.Add(rwmutexMaxReaders)
	if r >= rwmutexMaxReaders { // r >= 0 并且 < rwmutexMaxReaders 才是正常的（r 是持有写锁期间尝试获取读锁的 reader 数量）
		fatal("sync: Unlock of unlocked RWMutex")
	}
	// 如果有 reader 在等待写锁释放，那么唤醒这些 reader。
	for i := 0; i < int(r); i++ {
		runtime_Semrelease(&rw.readerSem, false, 0)
	}
	// 允许其他的 writer 继续进行。
	rw.w.Unlock()
}

写锁的实现总结：

获取写锁的时候，会将 readerCount 减去 rwmutexMaxReaders，这样就可以区分读锁和写锁了。
如果 readerCount 减去 rwmutexMaxReaders 之后不为 0，说明有 reader 正在使用读锁，那么就需要阻塞等待这些 reader 释放读锁。
释放写锁的时候，会将 readerCount 加上 rwmutexMaxReaders。
如果 readerCount 加上 rwmutexMaxReaders 之后大于 0，说明有 reader 正在等待写锁释放，那么就需要唤醒这些 reader。

TryRLock 和 TryLock

TryRLock 和 TryLock 的实现都很简单，都是尝试获取读锁或者写锁，如果获取不到就返回 false，获取到了就返回 true，这两个方法不会阻塞等待。

// TryRLock 尝试锁定 rw 以进行读取，并报告是否成功。
func (rw *RWMutex) TryRLock() bool {
	for {
		c := rw.readerCount.Load()
		// 有 goroutine 持有写锁
		if c < 0 {
			return false
		}
		// 尝试获取读锁
		if rw.readerCount.CompareAndSwap(c, c+1) {
			return true
		}
	}
}

// TryLock 尝试锁定 rw 以进行写入，并报告是否成功。
func (rw *RWMutex) TryLock() bool {
	// 写锁被占用
	if !rw.w.TryLock() {
		return false
	}
	// 读锁被占用
	if !rw.readerCount.CompareAndSwap(0, -rwmutexMaxReaders) {
		// 释放写锁
		rw.w.Unlock()
		return false
	}
	// 成功获取到锁
	return true
}

总结

RWMutex 使用起来比较简单，相比 Mutex 而言，它区分了读锁和写锁，可以提高并发性能。最后，总结一下本文内容：

RWMutex 有两种锁：读锁和写锁。
读锁可以被多个 goroutine 同时持有，写锁只能被一个 goroutine 持有。也就是可以并发读，但只能互斥写。
写锁被占用的时候，其他的读和写操作都会被阻塞。读锁被占用的时候，其他的写操作会被阻塞，但是读操作不会被阻塞。除非读操作发生在一个新的写操作之后。
RWMutex 包含以下几个方法：
- Lock：获取写锁，如果有其他的写锁或者读锁被占用，那么就会阻塞等待。
- Unlock：释放写锁。
- RLock：获取读锁，如果写锁被占用，那么就会阻塞等待。
- RUnlock：释放读锁。
也包含了两个非阻塞的方法：
- TryLock：尝试获取写锁，如果获取不到就返回 false，获取到了就返回 true。
- TryRLock：尝试获取读锁，如果获取不到就返回 false，获取到了就返回 true。
RWMutex 使用的注意事项跟 Mutex 差不多：
- 使用之后不能复制
- Unlock 之前需要有 Lock 调用，否则 panic，RUnlock 之前需要有 RLock 调用，否则 panic。
- 不要忘记使用 Unlock 和 RUnlock 释放锁。
RWMutex 的实现：
- 写锁还是使用 Mutex 来实现。
- 获取读锁和写锁的时候，如果获取不到都会阻塞等待，直到被唤醒。
- 获取写锁的时候，会将 readerCount 减去 rwmutexMaxReaders，这样就可以直到有写锁被占用。释放写锁的时候，会将 readerCount 加上 rwmutexMaxReaders。
- 获取写锁的时候，如果还有读操作未完成，那么这一次获取写锁只会等待这部分未完成的读操作完成。所有后续的操作只能等待这一次写锁释放。