并发编程读写锁的实现原理（RWMutex）

Posted 2021-12-03 了 凡

博主介绍：

– 我是了 凡，喜欢每日在简书上投稿日更的读书感悟笔名：三月_刘超。专注于 Go Web 后端，了解过一些Python、Java、算法、前端等领域。微信公众号【了凡银河系】期待你的关注，企鹅群号（798829931）。未来大家一起加油啊~

---

前言

Mutex为了保证读写共享资源的安全性。不管读还是写都通过Mutex来保证只有一个goroutine访问共享资源，这样就会浪费，例如写少读多的情况下，即使一段时间内没有写操作，大量并发的读访问也不得不在Mutex的保护下变成了串行访问，这个时候，使用Mutex，对性能的影响就比较大。这个时候我们就可以采用区分读写操作。

如果某个读操作的goroutine持有了锁，这样的话，其它操作goroutine就不必一直等待，而且可以并发的访问共享变量，这样就可以将串行的读变成并行读，提高读操作的性能。当写操作的goroutine持有锁的时候，它就是一个排外锁，其它的写操作和读操作的goroutine，需要阻塞等待持有这个锁的goroutine释放锁。

这一类并发读写问题叫作readers-writers问题，意思是，同时可能有多个读或多个写，但是只要有一个线程执行写操作，其它的线程不能执行读写操作。

而Go标准库的RWMutex(读写锁)就是用来解决这类readers-writers问题的。接下来会介绍读写锁的使用场景、实现原理以及容易掉入的坑，避免在实际开发中犯相同的错误。

---

文章目录

• 前言
• 什么是RWMutex？
• RWMutex的实现原理
• RLock/RUnlock的实现
• Lock
• Unlock

---

什么是RWMutex？

标准库中的RWMutex是一个reader/writer互斥锁。RWMutex在某一时刻只能由任意数量的reader持有，或者是只被单个的writer持有。

RWMutex的方法只有5个。

• Lock/Unlock:写操作时调用的方法。如果锁已经被reader或者writer持有，那么，Lock方法会一直阻塞，直到能获取锁;
  Unlock则是配对的释放锁的方法。

• Rlock/RUnlock：读操作时调用的方法。如果锁已经被writer持有的话，RLock方法会一直阻塞，直到能获取到锁，否则就直接返回；而RUnlock是reader释放锁的方法。

• RLocker：这个方法的作用是为读操作返回一个Locker接口的对象。它的Lock方法会调用RWMutex的RLock方法，它的Unlock方法会调用RWMutex的RUnlock方法。

RWMutex的零值是未加锁的状态，所以，当你使用RWMutex的时候，无论是声明变量，还是嵌入到其它的struct中，都不必显示地初始化。

假设计数器使用RWMutex保护共享资源。计数器的count ++操作是写操作，而获取count的值是读操作，这个场景很适合读写锁，因为读写锁可以并行执行，写操作时只允许一个线程执行，这正是readers-writers问题。

使用10个goroutine进行读操作，每读取一次，sleep1毫秒，同时，还有一个goroutine进行写操作，每一秒写一次，这是一个1writer-n reader 的读写场景，而且写操作还不是频繁（一秒一次）：

type Counter struct {
   mu sync.RWMutex
   count uint64
}

func main(){
   var counter Counter
   for i := 0; i < 10; i++ { // 10个reader
      go func() {
         for {
            counter.Count() // 计数器读操作
            time.Sleep(time.Millisecond)
         }
      }()
   }
   
   for {
      counter.Incr() // 计数操作
      time.Sleep(time.Second)
   }
}

// Incr 使用写锁保护
func (c *Counter) Incr()  {
   c.mu.Lock()
   c.count++
   c.mu.Unlock()
}

// Count 使用读锁保护
func (c *Counter) Count() uint64 {
   c.mu.RLocker()
   defer c.mu.RUnlock()
   return c.count
}

incr方法中count++，是一个写操作，使用Lock和Unlock进行保护。而Count方法是一个读取count值的一个操作，使用RLock/RUnlock方法进行保护。

incr方法每秒才调用一次，所以，writer竞争锁的频次是比较低的，而10个goroutine每毫秒都要执行一次查询，通过读写锁，并发的进行，可以极大提升计数器的性能。如果使用Mutex，性能就不会像读写锁这么好。因为多个reader并发读的时候，使用互斥锁导致了reader要排队读的情况，没有RWMutex并发读的性能好。

如果遇到可以明确区分reader和writer goroutine的场景，且有大量的并发读、少量的并发写，并且有强烈的性能需求，你就可以考虑使用读写锁RWMutex替换Mutex。

RWMutex的实现原理

RWMutex是很常见的并发原语，很多编程语言的库都提供了类似的并发类型。RWMutex一般都是基于互斥锁，条件变量（condition variables）或者信号量（semaphores）等并发原语来实现。Go标准库中的RWMutex是基于Mutex实现的。

readers-writers问题一般有三类，基于对读和写操作的优先级，读写锁的设计和实现也分成三类。

• Read-preferring：读优先的设计可以提供很高的并发性，但是，在竞争激烈的情况不可能导致写饥饿。这是因为，如果有大量的读，这种设计会导致只有所有的读都释放了锁之后，写才可能获取到锁。

• Writer-preferring：写优先的设计意味着，如果已经有一个writer在等待请求锁的话，它会阻止新来的请求锁的reader获取到锁，所以优先保证writer。当然，如果有一些reader已经请求了锁的话，新请求的writer也会等待已经存在的reader都释放锁之后才能获取。写优先级设计中的优先权是针对新来的请求而言的。这种设计主要避免了writer的饥饿问题。

• 不指定优先级：这种设计比较简单，不区分reader和writer优先级，某些场景下这种不指定优先级的设计反而更有效，因为第一类优先级会导致饥饿问题。

Go标准库中RWMutex设计是Write-preferring方案。一个正在阻塞的Lock调用会排除新的reader请求到锁。

RWMutex包含一个Mutex，以及四个辅助字段writerSem、readerSem、readerCount和readerWait：

type RWMutex struct{
    w   Mutex   // 互斥锁解决多个writer的竞争
    writerSem   uint32  // writer信号量
    readerSem   uint32   // reader信号量
    readerCount int32    // reader的数量
    readerWait  int32    // writer等待完成的reader的数量
}

const  rwmutexMaxReaders = 1 << 30

• 字段w：为writer的竞争锁而设计
• 字段readerCount：记录当前reader的数量（以及是否有writer竞争锁）；
• readerWait：记录writer请求锁时需要等待read完成的reader的数量；
• writerSem和readerSem：都是为了阻塞设计的信号量。

这里的常量rwmutexMaxReaders，定义了最大的reader数量。

RLock/RUnlock的实现

移除了race的等无关紧要的代码后的RLock和RUnlock方法：

func (rw *RWMutex) RLock(){
    if atomic.AddInt32($rw.readerCount, 1) < 0 {
        // rw.readerCount是负值的时候，意味着此时有writer等待请求锁，因为writer优先级
        rutime_SemacquireMutex(&rw.readerSem,false,0)
    }
}

func (rw *RWMutex) RUnlock() {
    if r := atomic.AddInt32(&rw.readerCount,-1);r < 0 {
        rw.rUnlockSlow(r) // 有等待的write
    }
}

func(rw *RWMuter) rUnlockSlow(r int32) {
    if atomic.AddInt32(&rw.readerWait, -1) == 0{
        // 最后一个reader了，writer终于有机会获得锁了
        runtime_Semrele(&)
    }
}

第2行是对reader计数加1。但是readerCount怎么还可能为负数呢？其实，这是因为，readerCount这个字段有双重含义：

• 没有writer竞争或持有锁时就，readerCount和我们正常理解的reader的计数是一样的;
• 但如果有writer竞争锁或者持有锁时，那么，readerCount不仅仅承担着reader的计数功能，还能够标识当前是否有writer竞争或持有锁，这种情况，请求锁的reader的处理进入第4行，阻塞等锁的释放。

调用RUnlock的时候，我们需要将Reader的计数减去1，因为reader的数量减少了一个。但是，第8行的AddInt32的返回值还有另外一个含义。如果它负值，就表示当前有writer竞争锁，在这种情况下，还会调用rUnlockSlow方法，检查是不是reader都释放读锁了，如果读锁都释放了，那么可以唤请求锁的writer了。

当一个或者多个reader持有锁的时候，竞争锁的writer会等待这些reader释放完，才可能持有这把锁。打个比方，在房地产行业中有条规矩叫做“买卖不破租凭”，意思是说，就算房东把房子卖了，新业主也不能把当前的租户赶走，而是等到租约结束后，才能接管房子。和RWMutex的设计是一样的。当writer请求锁的时候，是无法改变既有的reader持有锁的现实的，也不会强制这些
reader释放锁，它的优先权只是限定后来的reader不要和它抢。

rUnlockSlow将持有锁的reader计数减少1的时候，会检查既有的reaer是不是都已经释放了锁，如果都释放了锁，就会唤醒writer，让writer持有锁。

Lock

RWMutex是一个多writer多reader的读写锁，所以同时可能有多个writer和reader。那么，为了避免writer之间的竞争，RWMutex就会使用一个Mutex来保证writer的互斥。

一旦一个writer获得了内部的互斥锁，就会反转readerCount字段，把它从原本的正整数readerCount(>=0)修改为负数，让这个字段保持两个含义(既保存了reader的数量，又表示当前writer)。

第5行，还会记录当前活跃的reader数量，所谓活跃的reader，就是指持有读锁还没有释放的那些reader。

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)
    }
}

如果readerCount不是0，就说明当前有持有读锁的reader，RWMutex需要把这个当前readerCount赋值给readerWait字段保存下来(第7行)，同时，这个writer进入阻塞等待状态（第8行）。

每当一个reader释放锁的时候（调用RUnlock方法时），readerWait赋值给readerWait字段就减1，直到所有的活跃的reader都释放了读锁，才会唤醒这个writer。

Unlock

当一个writer释放锁的时候，它会再次反转readerCount字段。可以肯定的是，因为当前锁由writer持有，所以，readerCount字段是反转过来的，并且减去了rwmutexMaxReaders这个常数，变成了负数。所以，这里的反转方法就是给它增加rwmutexMaxReaders这个常数值。

既然writer要释放锁了，那么就需要唤醒之后新来的reader，不必再阻塞它们了，让它们开开心心地继续执行就好了。

在RWMutex的Unlock返回之前，需要把内部的互斥锁释放。释放完毕后，其他的writer才可以继续竞争这把锁。

func (rw *RWMutex) Unlock(){
    // 告诉reader没有活跃的writer了
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    
    // 唤醒阻塞的reader们
    for i := 0; i < int(r); i++ {
        rutime_Semrelease(&rw.readerSem, false, 0)
    }
    // 释放内部的互斥锁
    rw.w.Unlock()
}

可以看到，删除了race的处理和异常情况的检查，总体看来还是比较简单的。

注意：首先，要理解readerCount这个字段的含义以及反转方式。其次，还要注意字段的更改的内部互斥锁的顺序关系。再Lock方法中，是先获取内部互斥锁，才会修改的其他字段；而再Unlock方法中，是先修改的其他字段，才会释放内部互斥锁，这样才能保证字段的修改也受到互斥锁的保护。

---

这次就先讲到这里，如果想要了解更多的golang语言内容一键三连后序每周持续更新！

---