Go 并发编程之 RWMutex

Mutex 是用来保证只有一个 goroutine 访问共享资源，在大量的并发场景中，特别是读场景中，一个共享资源块只能让 goroutine 串行访问，这就导致了性能的影响，解决方法就是区分读写操作。 这样就可以将串行的读变成并行的读，用来提高读操作的性能。 Go 标准库 RWMutex （读写锁）就是用来解决 readers-writers 问题的。 RWMutex 标准库中的 RWMutex 是一个 reader/writer 互斥锁，RWMutex 在同一时间只能由 n 个 reader 持有，或者只能被单个的 writer 持有。 Lock/Unlock：写操作时调用的方法，若是被 reader 或者 writer 持有， Lock 会一直阻塞，直到可以获取到锁，Ulock 是释放锁； Rlock/RUnlock：读操作时低哦用的方法，如果已经被 writer 持有的话， Rlock 会一直阻塞，直到获取到锁，否者直接返回， RUlock 是 reader 释放锁的方法； RLocker：为读操作返回一个 Locker 接口的对象。 RWMutex 的零值是未加锁的状态，所以在使用 RWMutex 作为变量或者嵌入到 struct 中去，都没有必要进行显式的初始化。 实现原理 针对于 readers-writers 问题是基于对读和写操作的优先级，读写锁的设计分为三类： Read-preferring 读优先设计：并发效果好，但是在大量的并发场景下会导致写饥饿； Write-preferring 写优先设计：针对新请求而言，主要是避免了 writer 饥饿问题，也就是说同一时间有一个 reader 和 writer 等待获取锁，会优先给 writer； 不指定优先级：FIFO，不区分读写优先级，适用于某些特定的场景。 RWMutex 设计是 write-preferring 写优先设计。一个正在阻塞的 Lock 调用会排出新的 reader 请求到锁。 RWMutex 包含一个 Mutex，以及四个辅助字段 writerSem、readerSem、readerCount 和 readerWait： 1 2 3 4 5 6 7 8 9 type RWMutex struct { w Mutex // 互斥锁解决多个 writer 的竞争 writerSem uint32 // writer 信号量 readerSem uint32 // reader 信号量 readerCount int32 // reader 的数量，记录当前 reader 等待的数量 readerWait int32 // writer 等待完成的 reader的 数量 } const rwmutexMaxReaders = 1 << 30 RLock/RUlock 实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 func (rw *RWMutex) RLock() { // 对 reader 计数 +1，readerCount 会出现负数 // 1、没有 writer 竞争或者持有锁的时候，readerCount 充当计数器存在 // 2、如果有 writer 竞争锁或者持有锁时，那么，readerCount 不仅仅承担着 reader 的计数功能，还能够标识当前是否有 writer 竞争或持有锁 if atomic.AddInt32(&rw.readerCount, 1) < 0 { // rw.readerCount 是负值的时候，意味着此时有 writer 等待请求锁 // 因为writer优先级高，所以把后来的 reader 阻塞休眠 runtime_SemacquireMutex(&rw.readerSem, false, 0) } } func (rw *RWMutex) RUnlock() { // 将 reader 计数 -1 if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 { // 如果为 负数，代表着当前有 writer 在竞争锁，检查是不是所有的 reader 都将锁释放 // 若释放了就让 writer 获取到锁进行写操作 rw.rUnlockSlow(r) // 有等待的writer } } func (rw *RWMutex) rUnlockSlow(r int32) { // rUnlockSlow 将持有锁的 reader 计数 -1 的时候； // 会检查既有的 reader 是不是都已经释放了锁； // 如果都释放了锁，就会唤醒 writer，让 writer 持有锁。 if atomic.AddInt32(&rw.readerWait, -1) == 0 { // 最后一个reader了，writer终于有机会获得锁了 runtime_Semrelease(&rw.writerSem, false, 1) } } Lock / Unlock RWMutex 是一个多 writer 多 reader 的读写锁，所以同时可能有多个 writer 和 reader。那么，为了避免 writer 之间的竞争，RWMutex 就会使用一个 Mutex 来保证 writer 的互斥。 1 2 3 4 5 6 7 8 9 10 func (rw *RWMutex) Lock() { // 首先解决其他 writer 竞争问题 rw.w.Lock() // 反转 readerCount，告诉 reader 有 writer 竞争锁 r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders // 如果当前有 reader 持有锁，那么需要等待 if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 { runtime_SemacquireMutex(&rw.writerSem, false, 0) } } 一旦一个 writer 获得了内部的互斥锁，就会反转 readerCount 字段，把它从原来的正整数 readerCount(>=0) 修改为负数（readerCount-rwmutexMaxReaders），让这个字段保持两个含义（既保存了 reader 的数量，又表示当前有 writer）。 1 2 3 4 5 6 7 8 9 10 11 func (rw *RWMutex) Unlock() { // 告诉 reader 没有活跃的 writer 了 r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders) // 唤醒阻塞的 reader 们 for i := 0; i < int(r); i++ { runtime_Semrelease(&rw.readerSem, false, 0) } // 释放内部的互斥锁 rw.w.Unlock() } 当一个 writer 释放锁的时候，它会再次反转 readerCount 字段。可以肯定的是，因为当前锁由 writer 持有，所以，readerCount 字段是反转过的，并且减去了 rwmutexMaxReaders 这个常数，变成了负数。 所以，这里的反转方法就是给它增加 rwmutexMaxReaders 这个常数值。既然 writer 要释放锁了，那么就需要唤醒之后新来的 reader，不必再阻塞它们了，让它们开开心心地继续执行就好了。 在 RWMutex 的 Unlock 返回之前，需要把内部的互斥锁释放。释放完毕后，其他的 writer 才可以继续竞争这把锁。 RWMutex 常犯的三种错误 不可复制 重入导致死锁 释放未加锁的 RWMutex 关注微信公众号，第一时间获取最新内容，让我们一起变得更强！Debug客栈：订阅本站· 文章归档· 我的项目· 友情链接· 我的使用· 飞湾计划· 摄影展集· 我的主页· 知盈官网