关于golang:Golang-读写锁设计

8次阅读

共计 4709 个字符，预计需要花费 12 分钟才能阅读完成。

在《Go 精妙的互斥锁设计》一文中，咱们具体地解说了互斥锁的实现原理。互斥锁为了防止竞争条件，它只容许一个线程进入代码临界区，而因为锁竞争的存在，程序的执行效率会被升高。同时咱们晓得，只有多线程在共享资源中有写操作，才会引发竞态问题，只有资源没有发生变化，多线程读取雷同的资源就是平安的。因而，咱们引申出更细粒度的锁：读写锁。

读写锁是一种多读单写锁，分读和写两种锁，多个线程能够同时加读锁，然而写锁和写锁、写锁与读锁之间是互斥的。

读写锁对临界区的解决如上图所示。其中，t1 时刻，因为线程 1 已加写锁，线程 2 被互斥期待写锁的开释；t2 时刻，线程 2 已加读锁，线程 3 能够对其持续加读锁并进入临界区；t3 时刻，线程 3 加了读锁，线程 4 被互斥期待读锁的开释。

依据读写锁的性质，读者应该能猜到读写锁实用于读写明显的场景。依据优先级，能够分为读优先锁和写优先锁。读优先锁能容许最大并发，然而写线程可能被饿死；同理，写优先锁是优先服务写线程，这样读线程就可能被饿死。

相对而言，写锁饥饿的问题更为突出。因为读锁是共享的，如果以后临界区曾经加了读锁，后续的线程持续加读锁是没问题的，然而如果始终有读锁的线程加锁，那尝试加写锁的线程则会始终获取不到锁，这样加写锁的线程始终被阻塞，导致了写锁饥饿。

同时，因为多读锁共享，可能会有读者问：为什么不间接去掉读锁，在写操作线程进来时只加写锁就好了呢，这样岂不是很好实现了。情理很简略，如果以后临界区加了写锁，在写锁解开之前又有新的写操作线程进来，等到写锁开释，新的写操作线程又上了写锁。这种状况如果连续不断，那整个程序就只能执行写操作线程，读操作线程就活活被饿死了。

所以，为了防止饥饿问题，通用的做法是实现偏心读写锁，它将申请锁的线程用队列进行排队，保障了先入先出（FIFO）的准则进行加锁，这样就能无效地防止线程饥饿问题。

那 Go 语言的读写锁，对于饥饿问题，它是如何解决的呢？

本文代码版本为 Go 1.15.2。如下所示，sync.RWMutex构造体蕴含 5 个字段。

type RWMutex struct {
    w           Mutex  
    writerSem   uint32 
    readerSem   uint32 
    readerCount int32  
    readerWait  int32  
}

w 互斥锁sync.Mutex，用于互斥写操作。
writerSem 写操作 goroutine 阻塞期待信号量。最初一个阻塞写操作的读操作 goroutine 开释读锁时，会告诉阻塞的写锁 goroutine。
readerSem 读操作 goroutine 阻塞期待信号量。写锁 goroutine 开释写锁后，会告诉阻塞的读锁 goroutine。
readerCount 读操作 goroutine 数量。
readerWait 阻塞写操作 goroutine 的读操作 goroutine 数量。

对于互斥锁的几个字段，当初可能不好了解，先不必焦急，看完咱们对 sync.RWMutex 对外提供的四个办法接口解析，就天然明确了。

RLock()：加读锁
RUnlock()：解读锁
Lock()：加写锁
Unlock()：解写锁

上面，咱们来顺次剖析。

这里，须要阐明一下的是，为了更好了解代码逻辑，本文所有的代码块均去除了竞态检测的逻辑局部，即 if race.Enabled {} 办法块。

func (rw *RWMutex) RLock() {if atomic.AddInt32(&rw.readerCount, 1) < 0 {runtime_SemacquireMutex(&rw.readerSem, false, 0)
    }
}

atomic.AddInt32是一个原子性操作，其底层通过硬件指令 LOCK 实现封装（详情可见文章《同步原语的基石》）。rw.readerCount代表读操作 goroutine 数量，如果将其 +1，还小于 0，则通过用于同步库的 sleep 原语 runtime_SemacquireMutex 阻塞期待写锁开释。

简略地说，如果以后有写操作 goroutine 曾经进来了，则新来的读操作 goroutine 会被排队阻塞期待。然而，读者必定会感觉判断条件很奇怪，为什么 rw.readerCount 会是负值？不要急，下文会有答案。

当然，如果此时没有写锁，则仅仅将 rw.readerCount 数目加 1，而后间接退出，代表加读锁胜利。

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

将读操作 goroutine 数目 -1，如果其数目 r 大于等于 0，则间接退出，代表解读锁胜利。否则，带着以后处于负值的数目 r 进入以下 rUnlockSlow 逻辑

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 {runtime_Semrelease(&rw.writerSem, false, 1)
    }
}

如果 r+1==0，则证实在解读锁时，其实并没有读 goroutine 加读锁；rwmutexMaxReaders = 1 << 30，这代表读写锁所能接管的最大读操作 goroutine 数量。至于这里为什么r+1 == -rwmutexMaxReaders 也代表并没有 goroutine 加读锁，同样留在下文解答。在没有加读锁的锁上解读锁，会抛出异样并 panic。

rw.readerWait代表写操作被阻塞时，读操作 goroutine 数量。如果该值为 1，代表以后是最初一个阻塞写操作的 goroutine，则通过用于同步库的 wakeup 原语 runtime_Semrelease 唤醒阻塞的写操作 goroutine。

读者此时只看了加解读锁的代码，了解上会有艰难，不要急，咱们接着看加解写锁的逻辑。

func (rw *RWMutex) Lock() {rw.w.Lock()
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {runtime_SemacquireMutex(&rw.writerSem, false, 0)
    }
}

在加写锁时，首先会通过互斥锁加锁，这保障只会有一个写锁加锁胜利。当互斥锁加锁胜利之后，咱们就能看到 写操作是如何阻止读操作，读操作是如何感知到写操作的。

咱们曾经晓得 rw.readerCount 是代表读操作 goroutine 数量，如果在不存在写操作的状况下，每次加读锁，该值就会 +1，每次解读锁该值就会 -1，那么咱们能够正当地认为 rw.readerCount 的取值范畴是[0,rwmutexMaxReaders], 即最大反对 2^30 个并发读，最小是 0 个。

然而，在以后写操作 goroutine 加互斥锁胜利后，会通过原子操作 atomic.AddInt32 将readerCount减去 2^30，此时 readerCount 会变成负值，那么如果之后再有读操作 goroutine 加读锁时，能通过该负值晓得以后曾经有写锁了，从而阻塞期待。这里也解释了加读锁和解读锁两大节中留下的问题。最初，为了持有实在的读操作 goroutine 数目，再加回 2^30 即可。

这里须要分外留神的是：互斥锁加锁胜利并不意味着加写锁胜利。咱们须要晓得 读操作是如何阻止写操作，写操作是如何感知到读操作的。

r != 0 即代表以后读操作 goroutine 不为 0，这意味着写操作要期待排在后面的读操作完结后才算是加上写锁。写操作取得互斥锁后，通过 atomic.AddInt32 把rw.readerCount值拷贝到 rw.readerWait 中，用于标记排在写操作 goroutine 后面的读操作 goroutine 个数。通过用于同步库的 sleep 原语 runtime_SemacquireMutex 阻塞期待这些读操作完结。在解读锁小结中咱们晓得，读操作完结时，除了会递加 rw.readerCount，同时须要递加rw.readerWait 值，当 rw.readerWait 值变为 0 时就会唤醒阻塞的写操作 goroutine。

func (rw *RWMutex) Unlock() {r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {race.Enable()
        throw("sync: Unlock of unlocked RWMutex")
    }
    for i := 0; i < int(r); i++ {runtime_Semrelease(&rw.readerSem, false, 0)
    }
    rw.w.Unlock()}

在解写锁时，将负值的 rw.readerCount 变更为正值，解除对读锁的互斥，并唤醒 r 个因为写锁而阻塞的读操作 goroutine。最初，通过调用互斥锁的 Unlock 办法，解除对写锁的互斥。

到这里，咱们能够图解一下Go 是如何解决饥饿问题的。

假如 G1、G2、G3 是正在共享读的 goroutine，rw.readerCount值为 3。此时写操作 G4 进来，把 rw.readerCount 值变为了负值，同时它发现 rw.readerCount 不为 0，因而阻塞期待。然而在 G4 的期待期间又有新的读操作 G5、G6 和写操作 G7 进来。因为此时的 rw.readerCount 值为负，所以 G5 和 G6 是不能加读锁胜利的，会陷入阻塞期待。G7 因为 G4 加了互斥锁也会陷入期待。当排在写操作 G4 后面的最初一个读操作 G3 完结，G3 会唤醒 G4。当 G4 完结时，它将 G5、G6 和 G7 均唤醒。然而 G7 须要期待 G5 和 G6 退出（因为它在试图加写锁时，会发现 rw.readerCount 不为 0，会再次陷入阻塞期待）能力加写锁胜利。以此重复，保障了读写锁的绝对偏心，防止一方挨饿。

读写锁基于互斥锁，提供了更细粒度的管制，它实用于读写明显的场景，精确而言是读操作远多于写操作的状况。在多读少写的场景中，应用读写锁代替互斥锁能无效地进步程序运行效率。

读读共享、读写互斥和写写互斥。在优先级方面，偏袒读锁或者写锁要分几种状况。

锁闲暇，此时是齐全偏心的，谁先进来谁就能够上锁。
如果没有读操作，均是写操作，读写锁会进化成互斥锁，只有在互斥锁处于饥饿模式下才会偏心。
如果没有写操作，均是读操作，读操作均能够进来，读写锁进化成无锁设计（也并不是真正的无锁，因为加解锁均有原子操作 atomic.AddInt32 对读操作 goroutine 的统计）。
被加读锁时，写操作进来会被阻塞。在写操作阻塞期间，如果有读操作试图进来，它们也会被阻塞。当阻塞写操作的最初一个读操作解读锁时，它只会唤醒被阻塞的写操作，之后进来的读操作须要该写操作实现之后被唤醒。这些被唤醒的读操作会比新的写操作（能够是新来的，也能够是因互斥锁而被阻塞的）先拿到锁，期待这些读操作实现，新的写操作能力拿到写锁。

因为读写锁是基于互斥锁之上的设计，不可避免地多做了一些工作。因而，并不是说应用读写锁的收益肯定会比互斥锁高。在抉择何种锁时，须要综合考量读写操作的比例，临界区代码的耗时。性能比对的内容本文就不再探讨，读者可自行测试。

正文完