「Golang」sync.RWMutex源码讲解

什么是sync.RWMutex

上次写过了sync.Mutex的源代码解析，这回写一下他的扩展版本，sync.RWMutex（下称读写锁）的源代码解析，首先看一下读写锁的作用，如下述：

sync/rwmutex.go中
// A RWMutex is a reader/writer mutual exclusion lock.
// The lock can be held by an arbitrary number of readers or a single writer.

白话来讲，读写锁就是一个可以并发读但是不可以并发写的锁（此处有疑：不知应该称互斥锁还是称为锁），由于互斥锁的特性，会导致将所有goroutine（下称协程）串行化，从而影响整体程序运行的性能，如果写的数量大于读的数量时，性能损耗暂且可以忽略不计（建议不要忽略），但是当读的数量大于写的数量时，互斥锁对性能的影响是很大的，所以此刻我们需要采用读写锁来进行读操作和写操作的分开。

读写锁可以同时有多个协程进行对某个数据对某个数据进行读取，但是同一时间内只能有一个协程对其进行修改，这样就可以大大提高并发中读操作过多情况下互斥锁只能在同一时间内有一个协程访问这一痛点。

读写锁与互斥锁一样，禁止被复制，因为一旦对读写锁或互斥锁进行复制，连带复制的包括锁本身的状态，也就是说当对一个互斥锁或者读写锁复制时，当锁当前状态已经处于Lock状态时，被复制出来得到的锁也将会处于Lock状态，所以 锁禁止复制

字段解析

接下来看一下读写锁的源代码实现，首先看一下读写锁的结构体字段以及相关的常量字段的用处

本文所有源代码版本基于1.15

type RWMutex struct {
   
	w           Mutex  // 一个互斥锁的字段，用户进行写时加互斥锁
	writerSem   uint32 // 一个writer的信号量，类似互斥锁中的信号量
	readerSem   uint32 // 一个reader的信号量，类似互斥锁中的信号量
	readerCount int32  // 两种作用，1:标记有多少拿到读锁的reader，2:是否有writer需要竞争
	readerWait  int32  // writer需要等待读锁解锁的reader的数量
}
const rwmutexMaxReaders = 1 << 30 // 最大reader的上限。即最多有多少的reader同时能拿到读锁

在标准库中有很多的字段或者变量都是使用位移操作进行一个字段作为多个含义去使用，比如我的【Golang】【标准库】sync.Mutex源码解析这篇文章中互斥锁的state字段就一个字段进行位移操作后代表4个含义的代表，位移操作的性能本身就很快，这样的写法也提高了整体锁的性能，作者认为，未来的开发过程可以充分使用这种方法进行开发。

由于是读写锁，那么加锁解锁过程就不能像互斥锁一样只是单一的Lock和Unlock，读写锁的提供的操作有五个分别是：

Lock/Unlock：用于writer需要进行写操作时时调用的方法，如果调用时读锁已经被reader所持有，那么将会等待从未调用该方法到调用此方法时所有持有读锁的reader解锁后才会进行writer写锁获取，Unlock是其配对的解锁操作。并且通知从获取到写锁时之后新来的等待读锁的reader获取读锁。
RLock/RUnlock：用于reader进行读操作时调用的方法，当此时没有写锁被获取时，直接获取到读锁，当有写锁被获取时，等待写锁的释放后才会被唤醒并获取读锁，RUlock 是其相反的方法，并且当没有需要等待的读锁时，会通知等待获取写锁的writer进行写锁的获取。
RLocker：这个方法的作用是返回一个读锁的Locker对象，调用Lock和Unlock的时候会调用RLock和RUlock，个人认为这个方法可以构造一个只读锁。

RLock()

了解了读写锁提供的几个方法，接下来就开始分析源代码，首先分析的时读操作时的RLock方法
本次的源代码解析，我删除了所有if race.Enabled { //todo } 语句，因为其是判断是进行判断当前程序是否开启了race竞态检测模式的代码，即在运行go程序时是否采用go run race xxx.go这种进行进行竞态检测运行模式，所以进行省略

func (rw *RWMutex) RLock() {
   
	// 首先对读计数器进行+1 并且判断+1后的值是否小于0 如果小于0则代表当前有已经被获取的写锁
	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
   
		// 此时需要进行阻塞挂起，等待写锁的解锁
		runtime_SemacquireMutex(&rw.readerSem, false, 0)
	}
}

读锁加锁操作没什么好说的，主要就是针对readerCount字段的判断，如果其+1仍未负数时就代表此时此刻写锁已经被获取，即需要进行阻塞等待写锁的解锁。

RUnlock()

接下来看读锁的RUnlock操作：

func (rw *RWMutex) RUnlock() {
   
		// 将已经加锁的读锁数量-1，如果此时-1后小于0时，则代表
		// 1:有可能反复解锁，此时需要抛出panic
		// 2:有writer正在等待获取写锁
		if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
   
		rw.rUnlockSlow(r)
	}
}

func (rw *RWMutex) rUnlockSlow(r int32) {
   
	// 不可重复解锁，此时抛出panic
	if r+1 == 0 || r+1 == -rwmutexMaxReaders {
   
		race.Enable()
		throw("sync: RUnlock of unlocked RWMutex")
	}
	// 此时有一个writer正在等待获取写锁，
	// 如果当前解锁的reader是最后一个需要等待的读锁
	// 则唤醒等待读锁释放完的writer进行写锁的获取
	if atomic.AddInt32(&rw.readerWait, -1) == 0 {
   
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

读锁的解锁其实也是很简单的实现，就是去判断是否有正在等待的写锁，如果没有就直接返回，否则就进行readerWait字段的校验判断其是否是最后一个需要等待的读锁后唤醒等待读锁释放完的writer进行写锁的获取。

读的加锁于解锁操作都已经说完，接下来说一下写锁的加锁和解锁过程。

Lock()

func (rw *RWMutex) Lock() {
    
	// 先将Mutex字段进行加锁，以免有其他写锁操作或者其他操作破坏数据
	rw.w.Lock()
	// 将readerCount进行取反操作 这也是此字段除了标记reader数量的第二个功能，进行写锁标记
	// 即标记有writer需要竞争
	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
	// 此时将取反的r值交给readerWait代表仍需要等待释放锁的reader的数量
	// 如果该数量为0 那么代表不需要等待则直接获取写锁即可
	// 否则就将writer挂起阻塞直至RUlock唤醒
	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
   
		runtime_SemacquireMutex(&rw.writerSem, false, 0)
	}
 }

写锁的加锁过程必须先对整体的结构体的Mutex进行加锁，以免有其他的写操作同时对写锁的竞争导致data race。然后进行当前持有读锁的reader的数量进行取反，并且将其值交给readerWait用于标记需要等待释放锁的reader的数量，如果该字段不等于0则代表需要进行读锁解锁等待。当reader调用RUlock时会进行对此字段的-1并且判断，如果此字段为0时，则唤醒writer的阻塞，使得writer获取到写锁。

Unlock()

func (rw *RWMutex) Unlock() {
   
	// 写锁进行解锁时首先将加锁时取反的readerCount再次取反
	// 也就是解除当前有写锁正在竞争的标记
	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
	// 如果取反后这个值大于rwmutexMaxReaders 就代表重复解锁
	// 抛出panic
	if r >= rwmutexMaxReaders {
   
		race.Enable()
		throw("sync: Unlock of unlocked RWMutex")
	}
	// 解锁完毕后需要根据等待的readerCount的数量去依次唤醒这些reader 
	// 这些reader是在Lock后再次请求获取读锁的reader的数量
	for i := 0; i < int(r); i++ {
   
		runtime_Semrelease(&rw.readerSem, false, 0)
	}
	// 把写锁的互斥锁解锁，以便于其他writer进行写操作的竞争
	rw.w.Unlock()
}