環(huán)境：go 1.19.8

在讀多寫少的情況下，即使一段時間內(nèi)沒有寫操作，大量并發(fā)的讀訪問也不得不在Mutex的保護下變成串行訪問，這種情況下，使用Mutex，對性能影響比較大。
所以就要區(qū)分讀寫操作。如果某個讀操作的g持有了鎖，其他讀操作的g就不必等待了，可以并發(fā)的訪問共享變量，這樣就可以將串行的讀變成并行的讀，提高讀操作的性能?？衫斫鉃楣蚕礞i。

當寫操作的g持有鎖，它是一個排他鎖，不管其他的g是寫操作還是讀操作，都需要阻塞等待持有鎖的g釋放鎖。

什么是RWMutex？

reader/writer互斥鎖，在某一時刻只能由任意數(shù)量的reader持有，或者是只被單個writer持有。
RWMutex實現(xiàn)了5個方法：

• Lock/Unlock：寫操作時調(diào)用。如果鎖已經(jīng)被reader或者writer持有，那么，Lock方法會一直阻塞，直到能獲取到鎖；Unlock是對應(yīng)的釋放鎖方法
• RLock/RUnlock：讀操作時調(diào)用。如果鎖已經(jīng)被writer持有，RLock方法會一直阻塞，直到能獲取鎖，否則直接return；Rnlock是對應(yīng)的釋放鎖方法
• RLocker：這個方法的作用是為讀操作返回一個 Locker 接口的對象

案例：計數(shù)器，1writer n reader

使用場景

如果可以明確區(qū)分 reader 和 writer goroutine ，且有大量的并發(fā)讀，少量的并發(fā)寫，并且有強烈的性能要求，可以考慮使用讀寫鎖RWMutex替換Mutex

實現(xiàn)原理

RWMutex 是很常見的并發(fā)原語，很多編程語言的庫都提供了類似的并發(fā)類型。RWMutex
一般都是基于互斥鎖、條件變量（condition variables）或者信號量（semaphores）等
并發(fā)原語來實現(xiàn)。Go 標準庫中的 RWMutex 是基于 Mutex 實現(xiàn)的。
reader-writers 問題，一般有三類，基于對讀和寫操作的優(yōu)先級，讀寫鎖的設(shè)計和實現(xiàn)也分成三類

• Read-Preferring：讀優(yōu)先的設(shè)計可以提供很高的并發(fā)性。但在競爭激烈的情況下會導致寫?zhàn)囸I
• Write-Preferring：如果有一個writer在等待請求鎖，它會阻止新來請求鎖reader獲取到鎖，優(yōu)先保障writer。當然，如果reader已經(jīng)獲得鎖，新請求的writer也需要等待已持有鎖的reader釋放鎖。寫優(yōu)先級設(shè)計中的優(yōu)先權(quán)是針對新來的請求而言的。這種設(shè)計主要避免了 writer 的饑餓問題。
• 不指定優(yōu)先級：這種設(shè)計比較簡單，不區(qū)分 reader 和 writer 優(yōu)先級，某些場景下這種不指定優(yōu)先級的設(shè)計反而更有效，因為第一類優(yōu)先級會導致寫?zhàn)囸I，第二類優(yōu)先級可能會導致讀饑餓，這種不指定優(yōu)先級的訪問不再區(qū)分讀寫，大家都是同一個優(yōu)先級，解決了饑餓的問題。

Go 標準庫中的 RWMutex 設(shè)計是 Write-preferring 方案。一個正在阻塞的 Lock 調(diào)用
會排除新的 reader 請求到鎖。

源碼解析

上鎖解鎖流程以及數(shù)值變化情況

rwmutexMaxReaders 的數(shù)量被初始化為1<<30，理想中，寫鎖不會持續(xù)很久，不會導致readerCount 自動從負值自動+1回到正值。

RLock/RUnlock實現(xiàn)

type RWMutex struct {
	w           sync.Mutex // hold if there are pending writers
	writerSem   uint32     // 寫 阻塞信號
	readerSem   uint32     // 讀 阻塞信號
	readerCount int32      // 正在讀的調(diào)用者數(shù)量/ 當為負數(shù)時 表示有write持有鎖
	readerWait  int32      // writer持有鎖之前正等待解鎖的數(shù)量
}
const rwmutexMaxReaders = 1 << 30
func (rw *RWMutex) RLock() {
	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
		// 寫端 持有鎖， 讀端阻塞
		runtime_SemacquireMutex(&rw.readerSem, false, 0)
	}
}
func (rw *RWMutex) RUnlock() {
	if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
		rw.rUnlockSlow(r)
	}
}
func (rw *RWMutex) rUnlockSlow(r int32) {
	if r+1 == 0 || r+1 == -rwmutexMaxReaders {
		fatal("sync: RUnlock of unlocked RWMutex")
	}
	if atomic.AddInt32(&rw.readerWait, -1) == 0 {
		// 無讀者等待，喚醒寫端等待者
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

RLock

第11行，上讀鎖，首先對readerCount進行原子加1，如果小于0則表示存在寫鎖，直接阻塞。為什么readerCount會存在負值？這個要看readerCount除了在RLock中處理，還在哪里被處理了?？梢钥吹皆讷@取寫鎖時有響應(yīng)代碼。后面在解釋。如果原子加大于等于0，則表示獲取讀鎖成功。

RUnlock

第18行，讀解鎖，對readerCount進行原子減1，如果小于零，則表示存在活躍的reader(即當前獲得互斥鎖的寫鎖之前獲取到讀鎖權(quán)限的讀者數(shù)量)，readerWait 字段就減 1，直到所有的活躍的 reader 都釋放了讀鎖，才會喚醒這個 write

Lock/Unlock

func (rw *RWMutex) Lock() {
	// 1. 先嘗試獲取互斥鎖
	rw.w.Lock()
	// 2. 看是否有其他正持有鎖的讀者，有則阻塞
	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
		// rc - rwmutexMaxReaders + rwmutexMaxReaders > 0說明還有等待者, 寫端阻塞
		runtime_SemacquireMutex(&rw.writerSem, false, 0)
	}
}
func (rw *rwMutex) Unlock() {
	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
	if r >= rwmutexMaxReaders {
		fatal("sync: Unlock of unlocked RWMutex")
	}
	// 如果有等待的讀者，先喚醒
	for i := 0; i < int(r); i++ {
		runtime_Semrelease(&rw.readerSem, false, 0)
	}
	// 釋放互斥鎖
	rw.w.Unlock()
}

Lock

• 先獲取互斥鎖
• 成功獲取后，r=readerCount-rwmutexMaxReaders，得到的數(shù)值就是一個負數(shù)，在加上rwmutexReaders就表示寫鎖等待者的數(shù)量，此時，如果r不等于0，且readerWait+r!=0，則表示有讀等待者，寫鎖阻塞

我們知道，寫操作要等待讀操作結(jié)束后才可以獲得鎖，寫操作等待期間可能還有新的讀操作持續(xù)到來，如果寫操作等待所有讀操作結(jié)束，就會出現(xiàn)饑餓現(xiàn)象。然而，通過readerWait可完美解決這個問題。

寫操作到來時，會把readerCount值拷貝到readerWait中，用于標記排在寫操作之前到讀者個數(shù)。
當讀操作結(jié)束后，除了會遞減readerCount，還會遞減readerWait的值，當readerWait值變?yōu)?時會喚醒寫操作。

寫操作之后產(chǎn)生的讀操作會加入到readerCount中，阻塞知道寫鎖釋放。

Unlock

上面說過，寫鎖之后來的讀者會被阻塞，所以在寫鎖釋放之際，會看是否有需要喚醒的讀者，再釋放互斥鎖

場景討論

寫操作如何阻塞寫操作

讀寫鎖包含一個互斥鎖(Mutex)，寫鎖必須先獲取該互斥鎖，如果互斥鎖已被協(xié)程A獲取，意味者其他協(xié)程只能阻塞等待互斥鎖釋放

寫操作是如何阻塞讀操作

readerCount是個整型值，用于表示讀者數(shù)量，不考慮寫操作的情況下，每次獲取讀鎖，將該值加1，每次解鎖將其減1，所以readerCount的取值為[0, N]，最大可支持2^30個并發(fā)讀者。

當寫鎖定進行時，會先將readerCount -= rwmutextMaxReaders(2^30)，此時 readerCount負數(shù)。這時再有讀者到了，檢測到readerCount為負值，則表示有寫操作正在進行，后來到讀者阻塞等待。等待者的數(shù)量即 reaerCount + 2^30

讀操作是如何阻止寫操作的

寫操作時，會把readerCount的值拷貝到readerWait中，用于標記在寫操作前面讀者的個數(shù)，前面的寫鎖釋放后，會遞減readerCount，readerWait，當readerWait值變?yōu)?時喚醒寫操作

3個踩坑點

不可復制

rwmutex是由一個互斥鎖和四個輔助字段組成的，與互斥鎖一樣，讀寫鎖也是不能復制的。
一旦讀寫鎖被使用，它的字段就會記錄它當前的一些狀態(tài)，如果此時去復制這把鎖，就會把它的狀態(tài)也復制過去。但原來的鎖在釋放的時候，并不會修改復制出來的讀寫鎖，會導致復制出來的讀寫鎖狀態(tài)異常，可能永遠無法釋放鎖。

重入導致死鎖

讀寫鎖重入，或者遞歸調(diào)用，導致的死鎖情況很多

讀寫鎖內(nèi)部基于互斥鎖實現(xiàn)對writer并發(fā)控制，而互斥鎖本身就有重入問題，所以，writer重入調(diào)用Lock，會導致死鎖

func foo(l *sync.RWMutex) {
    fmt.Println("lock in foo")
    l.Lock()
    bar(l)
    l.Unlock()
}
func bar(l *sync.RWMutex) {
    fmt.Println("lock in bar")
    l.Lock()
    l.Unlock()
}
func main() {
    l := &sync.RWMutex{}
    foo(l)
}

2.當一個 writer 請求鎖的時候，如果已經(jīng)有一些活躍的 reader，它會等待這些活躍的reader 完成，才有可能獲取到鎖，但是，如果之后活躍的 reader 再依賴新的 reader 的話，這些新的 reader 就會等待 writer 釋放鎖之后才能繼續(xù)執(zhí)行，這就形成了一個環(huán)形依賴： writer 依賴活躍的 reader -> 活躍的 reader 依賴新來的 reader -> 新來的 reader依賴 writer。

func main() {
    var mu sync.RWMutex
    go func() {
        time.Sleep(200*time.Millisecond)
        mu.Lock()
        fmt.Println("Lock")
        time.Sleep(100*time.Millisend)
        mu.Unlock()
        fmt.Println("Unlock")
    }
    go func() {
        factorial(&mu, 10) // 計算10的階乘
    }
    select {}
}
// 
func factorial(m *sync.RWMutex, n int) {
    if n < 1 {
        return 0
    }
    fmt.Println("RLock")
    m.RLock()
    defer func() {
        fmt.Println("RUnlock")
        m.RUnlock()
    }
    time.Sleep(100*time.Millisecond)
    return factorial(m, n-1) * n
}

factorial 方法是一個遞歸計算階乘的方法，我們用它來模擬 reader。為了更容易地制造出死鎖場景，在這里加上了 sleep 的調(diào)用，延緩邏輯的執(zhí)行。這個方法會調(diào)用讀鎖（第 27
行），在第 33 行遞歸地調(diào)用此方法，每次調(diào)用都會產(chǎn)生一次讀鎖的調(diào)用，所以可以不斷地產(chǎn)生讀鎖的調(diào)用，而且必須等到新請求的讀鎖釋放，這個讀鎖才能釋放。同時，我們使用另一個 goroutine 去調(diào)用 Lock 方法，來實現(xiàn) writer，這個 writer 會等待200 毫秒后才會調(diào)用 Lock，這樣在調(diào)用 Lock 的時候，factoria 方法還在執(zhí)行中不斷調(diào)用
RLock。這兩個 goroutine 互相持有鎖并等待，誰也不會退讓一步，滿足了“writer 依賴活躍的reader -> 活躍的 reader 依賴新來的 reader -> 新來的 reader 依賴 writer”的死鎖條件，所以就導致了死鎖的產(chǎn)生。

釋放未加鎖的RWMutex

鎖都是成對出現(xiàn)的，Lock和RLock的多余調(diào)用會導致鎖沒有被釋放，可能會出現(xiàn)死鎖。
而Unlock和RUnlock多余調(diào)用會導致panic

參考

go中sync.RWMutex源碼解讀

到此這篇關(guān)于從源碼深入理解golang RWMutex讀寫鎖操作的文章就介紹到這了,更多相關(guān)go讀寫鎖RWMutex內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！