快捷導(dǎo)航

從源碼深入理解golang?RWMutex讀寫(xiě)鎖操作

更新時(shí)間：2023年05月05日 11:15:12 作者：JonPan

這篇文章主要介紹了從源碼深入理解golang?RWMutex讀寫(xiě)鎖操作,本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

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

當(dāng)寫(xiě)操作的g持有鎖，它是一個(gè)排他鎖，不管其他的g是寫(xiě)操作還是讀操作，都需要阻塞等待持有鎖的g釋放鎖。

什么是RWMutex？

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

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

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

使用場(chǎng)景

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

實(shí)現(xiàn)原理

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

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

Go 標(biāo)準(zhǔn)庫(kù)中的 RWMutex 設(shè)計(jì)是 Write-preferring 方案。一個(gè)正在阻塞的 Lock 調(diào)用
會(huì)排除新的 reader 請(qǐng)求到鎖。

源碼解析

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

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

RLock/RUnlock實(shí)現(xiàn)

type RWMutex struct {
	w           sync.Mutex // hold if there are pending writers
	writerSem   uint32     // 寫(xiě) 阻塞信號(hào)
	readerSem   uint32     // 讀 阻塞信號(hào)
	readerCount int32      // 正在讀的調(diào)用者數(shù)量/ 當(dāng)為負(fù)數(shù)時(shí) 表示有write持有鎖
	readerWait  int32      // writer持有鎖之前正等待解鎖的數(shù)量
}
const rwmutexMaxReaders = 1 << 30
func (rw *RWMutex) RLock() {
	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
		// 寫(xiě)端 持有鎖， 讀端阻塞
		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 {
		// 無(wú)讀者等待，喚醒寫(xiě)端等待者
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

RLock

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

RUnlock

第18行，讀解鎖，對(duì)readerCount進(jìn)行原子減1，如果小于零，則表示存在活躍的reader(即當(dāng)前獲得互斥鎖的寫(xiě)鎖之前獲取到讀鎖權(quán)限的讀者數(shù)量)，readerWait 字段就減 1，直到所有的活躍的 reader 都釋放了讀鎖，才會(huì)喚醒這個(gè) 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說(shuō)明還有等待者, 寫(xiě)端阻塞
		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ù)值就是一個(gè)負(fù)數(shù)，在加上rwmutexReaders就表示寫(xiě)鎖等待者的數(shù)量，此時(shí)，如果r不等于0，且readerWait+r!=0，則表示有讀等待者，寫(xiě)鎖阻塞

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

寫(xiě)操作到來(lái)時(shí)，會(huì)把readerCount值拷貝到readerWait中，用于標(biāo)記排在寫(xiě)操作之前到讀者個(gè)數(shù)。
當(dāng)讀操作結(jié)束后，除了會(huì)遞減readerCount，還會(huì)遞減readerWait的值，當(dāng)readerWait值變?yōu)?時(shí)會(huì)喚醒寫(xiě)操作。

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

Unlock

上面說(shuō)過(guò)，寫(xiě)鎖之后來(lái)的讀者會(huì)被阻塞，所以在寫(xiě)鎖釋放之際，會(huì)看是否有需要喚醒的讀者，再釋放互斥鎖

場(chǎng)景討論

寫(xiě)操作如何阻塞寫(xiě)操作

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

寫(xiě)操作是如何阻塞讀操作

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

當(dāng)寫(xiě)鎖定進(jìn)行時(shí)，會(huì)先將readerCount -= rwmutextMaxReaders(2^30)，此時(shí) readerCount負(fù)數(shù)。這時(shí)再有讀者到了，檢測(cè)到readerCount為負(fù)值，則表示有寫(xiě)操作正在進(jìn)行，后來(lái)到讀者阻塞等待。等待者的數(shù)量即 reaerCount + 2^30

讀操作是如何阻止寫(xiě)操作的

寫(xiě)操作時(shí)，會(huì)把readerCount的值拷貝到readerWait中，用于標(biāo)記在寫(xiě)操作前面讀者的個(gè)數(shù)，前面的寫(xiě)鎖釋放后，會(huì)遞減readerCount，readerWait，當(dāng)readerWait值變?yōu)?時(shí)喚醒寫(xiě)操作

3個(gè)踩坑點(diǎn)

不可復(fù)制

rwmutex是由一個(gè)互斥鎖和四個(gè)輔助字段組成的，與互斥鎖一樣，讀寫(xiě)鎖也是不能復(fù)制的。
一旦讀寫(xiě)鎖被使用，它的字段就會(huì)記錄它當(dāng)前的一些狀態(tài)，如果此時(shí)去復(fù)制這把鎖，就會(huì)把它的狀態(tài)也復(fù)制過(guò)去。但原來(lái)的鎖在釋放的時(shí)候，并不會(huì)修改復(fù)制出來(lái)的讀寫(xiě)鎖，會(huì)導(dǎo)致復(fù)制出來(lái)的讀寫(xiě)鎖狀態(tài)異常，可能永遠(yuǎn)無(wú)法釋放鎖。

重入導(dǎo)致死鎖

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

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

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.當(dāng)一個(gè) writer 請(qǐng)求鎖的時(shí)候，如果已經(jīng)有一些活躍的 reader，它會(huì)等待這些活躍的reader 完成，才有可能獲取到鎖，但是，如果之后活躍的 reader 再依賴新的 reader 的話，這些新的 reader 就會(huì)等待 writer 釋放鎖之后才能繼續(xù)執(zhí)行，這就形成了一個(gè)環(huán)形依賴： writer 依賴活躍的 reader -> 活躍的 reader 依賴新來(lái)的 reader -> 新來(lái)的 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) // 計(jì)算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 方法是一個(gè)遞歸計(jì)算階乘的方法，我們用它來(lái)模擬 reader。為了更容易地制造出死鎖場(chǎng)景，在這里加上了 sleep 的調(diào)用，延緩邏輯的執(zhí)行。這個(gè)方法會(huì)調(diào)用讀鎖（第 27
行），在第 33 行遞歸地調(diào)用此方法，每次調(diào)用都會(huì)產(chǎn)生一次讀鎖的調(diào)用，所以可以不斷地產(chǎn)生讀鎖的調(diào)用，而且必須等到新請(qǐng)求的讀鎖釋放，這個(gè)讀鎖才能釋放。同時(shí)，我們使用另一個(gè) goroutine 去調(diào)用 Lock 方法，來(lái)實(shí)現(xiàn) writer，這個(gè) writer 會(huì)等待200 毫秒后才會(huì)調(diào)用 Lock，這樣在調(diào)用 Lock 的時(shí)候，factoria 方法還在執(zhí)行中不斷調(diào)用
RLock。這兩個(gè) goroutine 互相持有鎖并等待，誰(shuí)也不會(huì)退讓一步，滿足了“writer 依賴活躍的reader -> 活躍的 reader 依賴新來(lái)的 reader -> 新來(lái)的 reader 依賴 writer”的死鎖條件，所以就導(dǎo)致了死鎖的產(chǎn)生。