一文掌握go的sync.RWMutex鎖

Posted on by

在簡略的說之前，首先要對RW鎖的結構有一個大致的瞭解

type RWMutex struct {
    w           Mutex  // 寫鎖互斥鎖，隻鎖寫鎖，和讀鎖無關
    writerSem   uint32 // sema鎖--用於“寫協程”排隊等待
    readerSem   uint32 // sema鎖--用於“讀協程”排隊等待
    readerCount int32  // 讀鎖的計數器
    readerWait  int32  // 等待讀鎖釋放的數量
}

這裡要額外說一句，writerSem和readerSem底層都是semaRoot，這個結構體有興趣可以瞭解下，他的用法有點類似於一個簡版的channel，很多地方把他的初始值設置為0，使得所有想獲取該sema鎖的協程都排隊等待，也就是說初始值為0的sema鎖，他本身起到的作用是成為一個協程等待隊列，就像沒有緩沖區的channel一樣。

好現在進入正題。本文是為瞭在面試中能快速口述RW鎖，並非為瞭完整解答RW鎖的機制。

前提：

readerCount這個參數非常重要

  • 為負數時：說明此鎖已經被寫協程占據，所有渴望加讀鎖的協程被阻塞在readerSem
  • 為正數時：正數的數值為當前持有該鎖的所有讀協程的數量總和，所有渴望加寫鎖的協程被阻塞在writerSem

讀寫鎖互斥性

  • 讀鎖是並發的，可以多個協程持有一把讀鎖。
  • 寫鎖是唯一的，互斥的，同一時刻隻能有一個寫協程擁有寫鎖
  • 讀鎖和寫鎖是互斥的，寫鎖生效時，是不能有讀鎖被獲取到，同樣，必須所有的讀鎖都被釋放，或者壓根沒有讀協程獲取讀鎖，寫鎖方可被獲取。

一個很重要的參數：const rwmutexMaxReaders = 1 << 30 ，rwmutexMaxReaders 非常大，意思是最多能有rwmutexMaxReaders（1 << 30  十進制為  4294967296）個協程同時持有讀鎖。

寫鎖上鎖場景：

首先分析寫鎖，因為讀鎖的很多操作是根據寫鎖來的，如果一上來就說讀鎖，很多東西沒法串起來

 func (rw *RWMutex) Lock() {
    // race.Enabled是官方的一些測試，性能檢測的東西，無需關心，這個隻在編譯階段才能啟用
	if race.Enabled {
		_ = rw.w.state
		race.Disable()
	}
	// First, resolve competition with other writers.
	rw.w.Lock()
	// Announce to readers there is a pending writer.
	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
	// Wait for active readers.
	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
		runtime_SemacquireMutex(&rw.writerSem, false, 0)
	}
	if race.Enabled {
		race.Enable()
		race.Acquire(unsafe.Pointer(&rw.readerSem))
		race.Acquire(unsafe.Pointer(&rw.writerSem))
	}
}

1.獲取寫鎖–沒有讀鎖等待

  • rw.w.Lock進行加鎖，阻塞後續的其他寫協程的鎖請求。
  • atomic.AddInt32進行原子操作，減去rwmutexMaxReaders，減成功才說明沒有並發問題，可以繼續下面的操作。然後再加上rwmutexMaxReaders，得到真正的readerCount的數值。
  • 此時還需要再次進行一個原子操作，把當前readerCount的值搬運到readerWait裡面，意思是當前要獲取寫鎖的協程需要等待的讀鎖的數量。
  • 此時readerCount隻有兩種情況，一種是0，一種是正數，因為隻有寫鎖上的時候才為負數，而上面的操作已經還原瞭加寫鎖之前的值，而w.Lock保證瞭不會有2個及以上的寫協程去同時操作
  • readerCount 如果是 0，加鎖成功。
  • 如果不為0則說明有讀鎖等待，詳見場景2

2.獲取寫鎖–有讀鎖等待

  • 接上面的判斷，如果readrCount不為0，說明前面有讀鎖正在運行，寫鎖需要等待所有讀鎖釋放才能獲取寫鎖，當前協程執行 runtime_SemacquireMutex 進入 waiterSem 的休眠隊列等待被喚醒

3.獲取寫鎖–前面已經有寫鎖瞭

後面的寫協程也調用 rw.w.Lock() 進行加鎖，因為前面有寫鎖已經獲取瞭w，所以後續的寫協程會因為獲取不到w，而進入到w的sema隊列裡面，w是一個mutex的鎖，mutex鎖裡是一個sema鎖，sema鎖因為沒有設置初始值，所以退化為一個隊列，而獲取不到w鎖的就會直接被阻塞在w的sema隊列裡，從而無法進行接下來的操作

寫鎖釋放鎖場景：

func (rw *RWMutex) Unlock() {
	if race.Enabled {
		_ = rw.w.state
		race.Release(unsafe.Pointer(&rw.readerSem))
		race.Disable()
	}
 
	// Announce to readers there is no active writer.
	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
	if r >= rwmutexMaxReaders {
		race.Enable()
		throw("sync: Unlock of unlocked RWMutex")
	}
	// Unblock blocked readers, if any.
	for i := 0; i < int(r); i++ {
		runtime_Semrelease(&rw.readerSem, false, 0)
	}
	// Allow other writers to proceed.
	rw.w.Unlock()
	if race.Enabled {
		race.Enable()
	}
}

1.釋放寫鎖–後面【沒有】讀鎖等待

  • 執行atomic.AddInt32進行原子操作，把已經為負值的readerCount還原為正數，此時已經算釋放瞭寫鎖
  • （此步驟不重要，就是個判錯）如果還原後的readerCount比rwmutexMaxReaders還大，這就是說明出錯瞭，直接throw彈出錯誤，throw這個方法是內部方法，對go來說就是panic瞭
  • 此場景因為沒有讀鎖等待，此時的readerCount為0，不會進入for循環，直接rw.w.Unlock釋放w鎖，允許其他寫協程加鎖，此時其他的寫協程會被從w裡的sema隊列裡喚醒

2.釋放寫鎖–後面【有】讀鎖等待

  • 接場景1，原子操作readerCount釋放寫鎖後，如果r是大於0，說明有讀鎖等待，for循環readerSem裡面所有的等待的讀協程，因為讀鎖是共享鎖，所以所有的讀協程都會獲取鎖並被喚醒
  • rw.w.Unlock釋放w鎖，允許其他寫協程加鎖，其他的寫協程會被從w裡的sema隊列裡喚醒

3.釋放寫鎖–後面有【寫鎖】等待

  • 上接場景2，當rw.w.Unlock釋放w鎖，其他的寫協程會被從w裡的sema隊列裡喚醒
  • 寫鎖釋放的時候，是先喚醒所有等待的讀鎖，再解除rw.w鎖，所以，並不會造成讀鎖的饑餓
  • 後面讀鎖再次對rw.w進行上鎖，重復上面所述寫鎖獲取鎖的場景

讀鎖上鎖場景：

func (rw *RWMutex) RLock() {
    // race.Enabled都是測試用的代碼，在閱讀源碼的時候可以跳過
	if race.Enabled {
		_ = rw.w.state
		race.Disable()
	}
    
	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
		// A writer is pending, wait for it.
		runtime_SemacquireMutex(&rw.readerSem, false, 0)
	}
	if race.Enabled {
		race.Enable()
		race.Acquire(unsafe.Pointer(&rw.readerSem))
	}
}

1.獲取讀鎖–此時沒有寫鎖.

最簡單的場景，協程對rw.readerCount進行原子操作加一，如果得到的結果為正數，說明獲取讀鎖成功。

2.獲取讀鎖–前方已經有寫鎖搶占瞭該鎖

  • 當協程對rw.readerCount進行原子加1操作的時候，發現加完，readerCount還是負數，說明在這個時間點以前，已經有協程獲取瞭寫鎖
  •  runtime_SemacquireMutex 方法將當前協程加入readerSem隊列，等待寫鎖釋放後被批量喚醒（寫鎖釋放會一次性放出所有的堆積的讀協程）

3.獲取讀鎖–前方有寫鎖搶已經被搶占，後方有寫鎖等待

  • 寫鎖在獲取的時候，對RWMutex.w進行加鎖，是獨占鎖，如果前方一個寫鎖已經得到瞭鎖正在處理業務，那麼後方的寫鎖進來就會發現加不上鎖，直接在rw.w.lock階段就阻塞瞭，後面的邏輯是無法繼續運行的，所以進入不瞭writerSem，它隻會進入到w這個mutex鎖的sema隊列裡，讀鎖則進入休眠隊列readerSem

讀鎖釋放鎖場景：

func (rw *RWMutex) RUnlock() {
	if race.Enabled {
		_ = rw.w.state
		race.ReleaseMerge(unsafe.Pointer(&rw.writerSem))
		race.Disable()
	}
	if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
		// Outlined slow-path to allow the fast-path to be inlined
		rw.rUnlockSlow(r)
	}
	if race.Enabled {
		race.Enable()
	}
}

1.釋放讀鎖–後方沒有寫鎖等待

  • atomic.AddInt32 進行原子操作，讓readerCount 減1，操作後，如果readerCount 大於0，說明後方是沒有寫鎖等待的，釋放鎖後整個流程就結束瞭

2.釋放讀鎖–後方有寫鎖等待

  • 原子操作eaderCount 減1後，發現eaderCount是小於0的，此時說明已經有等待寫鎖的協程在嘗試獲取寫鎖。執行 rw.rUnlockSlow(r) 。               
func (rw *RWMutex) rUnlockSlow(r int32) {
	if r+1 == 0 || r+1 == -rwmutexMaxReaders {
		race.Enable()
		throw("sync: RUnlock of unlocked RWMutex")
	}
	// A writer is pending.
	if atomic.AddInt32(&rw.readerWait, -1) == 0 {
		// The last reader unblocks the writer.
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

這裡是有個前提的，上面提到（詳見上面的獲取寫鎖的場景1），如果寫協程進來想加寫鎖，需要把它需要等待的讀鎖數量從readerCount裡賦值給readerWait。當它等待的讀鎖釋放後，就需要用rUnlockSlow方法對readerWait進行減1，如果readWait == 0 ，說明這是最後一個需要等待的讀鎖也釋放瞭，釋放後就通知該寫鎖可以被喚醒瞭，鎖給你瞭。

到此這篇關於go的sync.RWMutex鎖的文章就介紹到這瞭,更多相關go的sync.RWMutex鎖內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：