提示

本文代码基于 go1.17.13，src/sync/rwmutex.go

一、简述

• RWMutex 是一个读写互斥锁，锁可以被多个读 goroutine 同时拥有或被单个写 goroutine 拥有，但不能同时被读写 goroutine 拥有，即：写写互斥、写读互斥，只有读读是不互斥的；
• RWMutex 的零值是解锁的互斥锁，在首次使用后不得复制；
• RWMutex 适合读多写少的情况；

二、基本原理

• 写 goroutine 获取锁后会阻止其他写 goroutine 继续获得锁，这是通过互斥锁实现的，同时还会阻止其他读 goroutine 获得锁，此时读 goroutine 只能阻塞等待，直到该写 goroutine 释放锁后来唤醒读 goroutine，此时再来写 goroutine 尝试获取锁，只能排在这些读 goroutine 之后；
• 读 goroutine 获取锁后其他读 goroutine 能继续获得锁，会阻止写 goroutine 获得锁，此时写 goroutine 只能阻塞等待，并阻止后续的读 goroutine 获得锁，直到这些在读 goroutine 释放锁后唤醒写 goroutine，此时再来读 goroutine ，只能排在这些写 goroutine 之后；

三、基本用法

var m sync.RWMutex
var val int

func GetVal() int {
	m.RLock()
	defer m.RUnlock()
	return val
}

func SetVal(v int) {
	m.Lock()
	val = v
	m.Unlock()
}

• 分别可以在不同的 goroutine 中调用上述 GetVal()函数和SetVal(v int)，用于获取或修改 val变量；

四、源码解读

1, RWMutex

type RWMutex struct {
	w           Mutex  // 互斥锁
	writerSem   uint32 // 信号量，写等待读
	readerSem   uint32 // 信号量，读等待写
	readerCount int32  // 正在执行读操作的 goroutine 数量
	readerWait  int32  // 写操作被阻塞时等待的读操作个数
}

• w 复用互斥锁，用于保证只有一个写 goroutine 获得锁；
• writerSem，信号量，写 goroutine 通过该信号量阻塞等待读 goroutine 唤醒；
• readerSem，信号量，读 goroutine 通过该信号量阻塞等待写 goroutine 唤醒；

2, Lock()

点击查看

// Lock 锁定 rw 用于写入，如果锁已锁定以进行读取或写入，则 Lock 将被阻塞，直到锁可用
func (rw *RWMutex) Lock() {
	// 通过互斥锁加写锁，如果已经有其他 goroutine 加上了写锁该 goroutine 就只能阻塞或自旋等待
	// 加上写锁后，互斥锁也会阻止其他 goroutine 继续加锁，其他 goroutine 只能阻塞或自旋等待
	rw.w.Lock()
	// 加上写锁后，将读操作的数量置为负数，用于阻塞继续添加读锁
	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
		// 当读操作的数量不为 0 且 读操作等待加读锁的数量不为 0 ，则将当前 goroutine 阻塞
		// 即使已经获得了互斥锁（能够阻止后续写操作继续获得互斥锁）
		// 直到等待加读锁的 readerWait 为 0 后被唤醒
		runtime_SemacquireMutex(&rw.writerSem, false, 0)
	}
}

• 写锁复用互斥锁，加锁成功，能够阻止后续写 goroutine 继续添加写锁；加锁不成功，则阻塞等待，说明有其他 goroutine 已经加锁成功还在使用中；
• 加锁成功后，将读操作的数量置为负数，能够在 RLock 时阻塞继续添加读锁。并将读操作数添加到readerWait，不为 0 则表示当前还有读操作没有完成，需要阻塞等待，直到 readerWait 为 0 后被唤醒；

3, UnLock()

点击查看

func (rw *RWMutex) Unlock() {
	// 将 readerCount 从负的变为正的，以提示读 goroutine 互斥锁已经释放掉，RLock 时可以加锁成功
	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
	if r >= rwmutexMaxReaders {
		race.Enable()
		// 如果没有锁定就释放互斥锁，则直接抛出错误
		throw("sync: Unlock of unlocked RWMutex")
	}
	for i := 0; i < int(r); i++ {
		// 逐一唤醒阻塞等待中的读 goroutine，可以加读锁
		runtime_Semrelease(&rw.readerSem, false, 0)
	}
	// 释放互斥锁，允许其他读 goroutine 继续获取互斥锁
	rw.w.Unlock()
}

• 与互斥锁一样，锁定的 RWMutex 不与特定的 goroutine 相关联，一个 goroutine 可以 RLock（锁定）一个 RWMutex，然后安排另一个 goroutine 来 RUnlock（解锁）它；
• 将 readerCount 从负的变为正的，以提示读 goroutine 互斥锁已经释放掉，RLock 时可以加锁成功，如果没有锁定就释放互斥锁，则直接抛出错误；
• 逐一唤醒阻塞等待中的读 goroutine，可以加读锁，释放互斥锁，允许其他读 goroutine 继续获取互斥锁；
• 先唤醒阻塞等待中的读 goroutine 再释放互斥锁，能够让读 goroutine 先加读锁，再让写 goroutine 加互斥锁，避免读 goroutine 饿死；

4, RLock()

点击查看

func (rw *RWMutex) RLock() {
	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
		// 有写 goroutine 获得了锁，该读 goroutine 阻塞等待被唤醒
		runtime_SemacquireMutex(&rw.readerSem, false, 0)
	}
}

• RLock 锁定 rw 进行读取，它不应用于递归读锁定，被阻塞的 Lock 调用会阻止新读取器获取 RWMutex
• 直接让正在执行读操作的 goroutine 数量 readerCount 加一，如果readerCount为负数，说明当前已经有写 goroutine 获得了互斥锁，此时该 读 goroutine 阻塞等待被唤醒；
• 当调用上述 Lock 函数后，读操作完成，唤醒读 goroutine；

5, RUnLock()

点击查看

func (rw *RWMutex) RUnlock() {
	// 读者数量减一，小于 0，则有阻塞的写操作，便将 写操作被阻塞时等待的读操作个数 减一，为 0 时，便可以将写操作唤醒
	if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
		// 将 readerCount 减一，如果此时为负数，说明有写 goroutine 正在调用 Lock 尝试获取互斥锁
		rw.rUnlockSlow(r)
	}
}
func (rw *RWMutex) rUnlockSlow(r int32) {
	if r+1 == 0 || r+1 == -rwmutexMaxReaders {
		// 未加读锁便解锁，抛出错误
		race.Enable()
		throw("sync: RUnlock of unlocked RWMutex")
	}
	if atomic.AddInt32(&rw.readerWait, -1) == 0 {
		// 将 readerWait 减一，当等待的读操作个数为 0 时，将阻塞的写操作唤醒
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

• RUnlock 撤消单个 RLock 调用，不会影响其他正在读的 goroutine 。如果 rw 在进入 RUnlock 时未被锁定以读取，则为运行时错误。
• 直接让正在执行读操作的 goroutine 数量 readerCount 减一，如果readerCount大于等于 0，则直接返回；如果小于 0 ，进入慢解锁路径；
• 小于 0 ，说明此时有写 goroutine 获取互斥锁后被阻塞等待了，需要将readerWait减一；
• 如果readerWait为 0，则需要唤醒被阻塞的写 goroutine；

6, Locker

func (rw *RWMutex) RLocker() Locker {
	return (*rlocker)(rw)
}

type rlocker RWMutex
func (r *rlocker) Lock()   { (*RWMutex)(r).RLock() }
func (r *rlocker) Unlock() { (*RWMutex)(r).RUnlock() }

• RLocker 返回一个 Locker 接口，该接口通过调用 rw.RLock 和 rw.RUnlock.来实现 Lock 和 Unlock 方法