Go語(yǔ)言讀寫鎖RWMutex的源碼分析

第Go語(yǔ)言讀寫鎖RWMutex的源碼分析目錄前言RWMutex總覽深入源碼數(shù)據(jù)結(jié)構(gòu)RLock()RUnlock()Lock()Unlock()常見問題實(shí)戰(zhàn)一下

前言

在前面兩篇文章中初見GoMutex、GoMutex源碼詳解，我們學(xué)習(xí)了Go語(yǔ)言中的Mutex，它是一把互斥鎖，每次只允許一個(gè)goroutine進(jìn)入臨界區(qū)，可以保證臨界區(qū)資源的狀態(tài)正確性。但是有的情況下，并不是所有g(shù)oroutine都會(huì)修改臨界區(qū)狀態(tài)，可能只是讀取臨界區(qū)的數(shù)據(jù)，如果此時(shí)還是需要每個(gè)goroutine拿到鎖依次進(jìn)入的話，效率就有些低下了。例如房間里面有一幅畫，有人想修改，有人只是想看一下，完全可以放要看的一部分人進(jìn)去，等他們看完再讓修改的人進(jìn)去修改，這樣既提高了效率，也保證了臨界區(qū)資源的安全?？春托薷?，對(duì)應(yīng)的就是讀和寫，本篇文章我們就一起來學(xué)習(xí)下Go語(yǔ)言中的讀寫鎖sync.RWMutex。

說明：本文中的示例，均是基于Go1.1764位機(jī)器

RWMutex總覽

RWMutex是一個(gè)讀/寫互斥鎖，在某一時(shí)刻只能由任意數(shù)量的reader持有或者一個(gè)writer持有。也就是說，要么放行任意數(shù)量的reader，多個(gè)reader可以并行讀；要么放行一個(gè)writer，多個(gè)writer需要串行寫。

RWMutex對(duì)外暴露的方法有五個(gè)：

RLock()：讀操作獲取鎖，如果鎖已經(jīng)被writer占用，會(huì)一直阻塞直到writer釋放鎖；否則直接獲得鎖；RUnlock()：讀操作完畢之后釋放鎖；Lock()：寫操作獲取鎖，如果鎖已經(jīng)被reader或者writer占用，會(huì)一直阻塞直到獲取到鎖；否則直接獲得鎖；Unlock()：寫操作完畢之后釋放鎖；RLocker()：返回讀操作的Locker對(duì)象，該對(duì)象的Lock()方法對(duì)應(yīng)RWMutex的RLock()，Unlock()方法對(duì)應(yīng)RWMutex的RUnlock()方法。

一旦涉及到多個(gè)reader和writer，就需要考慮優(yōu)先級(jí)問題，是reader優(yōu)先還是writer優(yōu)先：

reader優(yōu)先：只要有reader要進(jìn)行讀操作，writer就一直等待，直到?jīng)]有reader到來。這種方式做到了讀操作的并發(fā)，但是如果reader持續(xù)到來，會(huì)導(dǎo)致writer饑餓，一直不能進(jìn)行寫操作；writer優(yōu)先：只要有writer要進(jìn)行寫操作，reader就一直等待，直到?jīng)]有writer到來。這種方式提高了寫操作的優(yōu)先級(jí)，但是如果writer持續(xù)到來，會(huì)導(dǎo)致reader饑餓，一直不能進(jìn)行讀操作；沒有優(yōu)先級(jí)：按照先來先到的順序，沒有誰(shuí)比誰(shuí)更優(yōu)先，這種相對(duì)來說會(huì)更公平。

我們先來看下RWMutex的運(yùn)行機(jī)制，就可以知道它的優(yōu)先級(jí)是什么了。

可以想象RWMutex有兩個(gè)隊(duì)伍，一個(gè)是包含所有reader和你獲得準(zhǔn)入權(quán)writer的隊(duì)列A，一個(gè)是還沒有獲得準(zhǔn)入權(quán)writer的隊(duì)列B。

隊(duì)列A最多只允許有一個(gè)writer，如果有其他writer，需要在隊(duì)列B等待；當(dāng)一個(gè)writer到了隊(duì)列A后，只允許它之前的reader執(zhí)行讀操作，新來的reader需要在隊(duì)列A后面排隊(duì)；當(dāng)前面的reader執(zhí)行完讀操作之后，writer執(zhí)行寫操作；writer執(zhí)行完寫操作后，讓后面的reader執(zhí)行讀操作，再喚醒隊(duì)列B的一個(gè)writer到隊(duì)列A后面排隊(duì)。

初始時(shí)刻隊(duì)列A中writerW1前面有三個(gè)reader，后面有兩個(gè)reader，隊(duì)列B中有兩個(gè)writer

RWMutex運(yùn)行示例：初始時(shí)刻

并發(fā)讀多個(gè)reader可以同時(shí)獲取到讀鎖，進(jìn)入臨界區(qū)進(jìn)行讀操作；writerW1在隊(duì)列A中等待，同時(shí)又來了兩個(gè)reader，直接在隊(duì)列A后面排隊(duì)

RWMutex運(yùn)行示例：并發(fā)讀

寫操作W1前面所有的reader完成后，W1獲得鎖，進(jìn)入臨界區(qū)操作

RWMutex運(yùn)行示例：寫操作

獲得準(zhǔn)入權(quán)W1完成寫操作退出，先讓后面排隊(duì)的reader進(jìn)行讀操作，然后從隊(duì)列B中喚醒W2到隊(duì)列A排隊(duì)。W2從隊(duì)列B到隊(duì)列A的過程中，R8先到了隊(duì)列A，因此R8可以執(zhí)行讀操作。R9、R10、R11在W2之后到的，所以在后面排隊(duì)；新來的W4直接在隊(duì)列B排隊(duì)。

RWMutex運(yùn)行示例：獲得準(zhǔn)入權(quán)

從上面的示例可以看出，RWMutex可以看作是沒有優(yōu)先級(jí)，按照先來先到的順序去執(zhí)行，只不過是多個(gè)reader可以并行去執(zhí)行罷了。

深入源碼

數(shù)據(jù)結(jié)構(gòu)

type

RWMutex

struct

{

w

Mutex

//

控制

writer

在

隊(duì)列B

排隊(duì)

writerSem

uint32

//

寫信號(hào)量，用于等待前面的

reader

完成讀操作

readerSem

uint32

//

讀信號(hào)量，用于等待前面的

writer

完成寫操作

readerCount

int32

//

reader

的總數(shù)量，同時(shí)也指示是否有

writer

在隊(duì)列A

中等待

readerWait

int32

//

隊(duì)列A

中

writer

前面

reader

的數(shù)量

//

允許最大的

reader

數(shù)量

const

rwmutexMaxReaders

=

1

30

上述中的幾個(gè)變量，比較特殊的是readerCount，不僅表示當(dāng)前所有reader的數(shù)量，同時(shí)表示是否有writer在隊(duì)列A中等待。當(dāng)readerCount變?yōu)樨?fù)數(shù)時(shí)，就代表有writer在隊(duì)列A中等待了。

當(dāng)有writer進(jìn)入隊(duì)列A后，會(huì)將readerCount變?yōu)樨?fù)數(shù)，即readerCount=readerCount-rwmutexMaxReaders，同時(shí)利用readerWait變量記錄它前面有多少個(gè)reader；如果有新來的reader，發(fā)現(xiàn)readerCount是負(fù)數(shù)，就會(huì)直接去后面排隊(duì)；writer前面的reader在釋放鎖時(shí)，會(huì)將readerCount和readerWait都減一，當(dāng)readerWait==0時(shí)，表示writer前面的所有reader都執(zhí)行完了，可以讓writer執(zhí)行寫操作了；writer執(zhí)行寫操作完畢后，會(huì)將readerCount再變回正數(shù)，readerCount=readerCount+rwmutexMaxReaders。

舉例：假設(shè)當(dāng)前有兩個(gè)reader，readerCount=2；允許最大的reader數(shù)量為10

當(dāng)writer進(jìn)入隊(duì)列A時(shí)，readerCount=readerCount-rwmutexMaxReaders=-8，readerWait=readerCount=2如果再來3個(gè)reader，readerCount=readerCount+3=-5獲得讀鎖的兩個(gè)reader執(zhí)行完后，readerCount=readerCount-2=-7，readerWait=readerWait-2=0，writer獲得鎖writer執(zhí)行完后，readerCount=readerCount+rwmutexMaxReaders=3，當(dāng)前有3個(gè)reader

RLock()

reader執(zhí)行讀操作之前，需要調(diào)用RLock()獲取鎖

func

(rw

*RWMutex)

RLock()

{

//

reader

加鎖，將

readerCount

加一，表示多了個(gè)

reader

if

atomic.AddInt32(rw.readerCount,

1)

0

{

//

如果

readerCount0，說明有

writer

在自己前面等待，排隊(duì)等待讀信號(hào)量

runtime_SemacquireMutex(rw.readerSem,

false,

0)

RUnlock()

reader執(zhí)行完讀操作后，調(diào)用RUnlock()釋放鎖

func

(rw

*RWMutex)

RUnlock()

{

//

reader

釋放鎖，將

readerCount

減一，表示少了個(gè)

reader

if

r

:=

atomic.AddInt32(rw.readerCount,

-1);

r

0

{

//

如果readerCount0，說明有

writer

在自己后面等待，看是否要讓

writer

運(yùn)行

rw.rUnlockSlow(r)

func

(rw

*RWMutex)

rUnlockSlow(r

int32)

{

//

將

readerWait

減一，表示前面的

reader

少了一個(gè)

if

atomic.AddInt32(rw.readerWait,

-1)

==

0

{

//

如果

readerWait

變?yōu)榱?，那么自己就是最后一個(gè)完成的

reader

//

釋放寫信號(hào)量，讓writer運(yùn)行

runtime_Semrelease(rw.writerSem,

false,

1)

Lock()

writer執(zhí)行寫操作之前，調(diào)用Lock()獲取鎖

func

(rw

*RWMutex)

Lock()

{

//

利用互斥鎖，如果前面有writer，那么就需要等待互斥鎖，即在隊(duì)列B中排隊(duì)等待；如果沒有，可以直接進(jìn)入

隊(duì)列A排隊(duì)

rw.w.Lock()

//

atomic.AddInt32(rw.readerCount,

-rwmutexMaxReaders)

將

readerCount

變成了負(fù)數(shù)

//

再加

rwmutexMaxReaders，相當(dāng)于

r

=

readerCount，r

就是

writer

前面的

reader

數(shù)量

r

:=

atomic.AddInt32(rw.readerCount,

-rwmutexMaxReaders)

+

rwmutexMaxReaders

//

如果

r!=

0

，表示自己前面有

reader，那么令

readerWait

=

r，要等前面的

reader

運(yùn)行完

if

r

!=

0

atomic.AddInt32(rw.readerWait,

r)

!=

0

{

runtime_SemacquireMutex(rw.writerSem,

false,

0)

Lock()和RUnlock()是會(huì)并發(fā)進(jìn)行的：

如果Lock()將readerCount變?yōu)樨?fù)數(shù)后，假設(shè)r=3，表示加入的那一刻前面有三個(gè)reader，還沒有賦值readerWaitCPU就被強(qiáng)占了，readerWait=0；假設(shè)此時(shí)三個(gè)reader的RUnlock()會(huì)進(jìn)入到rUnlockSlow()邏輯，每個(gè)reader都將readerWait減一，readerWait會(huì)變成負(fù)數(shù)，此時(shí)不符合喚醒writer的條件；三個(gè)reader運(yùn)行完之后，此時(shí)readerWait=-3，Lock()運(yùn)行到atomic.AddInt32(rw.readerWait,r)=-3+3=0，也不會(huì)休眠，直接獲取到鎖，因?yàn)榍懊娴膔eader都運(yùn)行完了。

這就是為什么rUnlockSlow()要判斷atomic.AddInt32(rw.readerWait,-1)==0以及Lock()要判斷atomic.AddInt32(rw.readerWait,r)!=0的原因。

Unlock()

writer執(zhí)行寫操作之后，調(diào)用Lock()釋放鎖

func

(rw

*RWMutex)

Unlock()

{

//

將

readerCount

變?yōu)檎龜?shù)，表示當(dāng)前沒有

writer

在隊(duì)列A

等待了

r

:=

atomic.AddInt32(rw.readerCount,

rwmutexMaxReaders)

//

將自己后面等待的

reader

喚醒，可以進(jìn)行讀操作了

for

i

:=

0;

i

int(r);

i++

{

runtime_Semrelease(rw.readerSem,

false,

0)

//

釋放互斥鎖，如果隊(duì)列B有writer，相當(dāng)于喚醒一個(gè)來隊(duì)列A排隊(duì)

rw.w.Unlock()

writer對(duì)readerCount一加一減，不會(huì)改變整體狀態(tài)，只是用正負(fù)來表示是否有writer在等待。當(dāng)然，如果在writer將readerCount變?yōu)樨?fù)數(shù)后，來了很多reader，將readerCount變?yōu)榱苏龜?shù)，此時(shí)reader在writer沒有釋放鎖的時(shí)候就獲取到鎖了，是有問題的。但是rwmutexMaxReaders非常大，可以不考慮這個(gè)問題。

常見問題

不可復(fù)制

和Mutex一樣，RWMutex也是不可復(fù)制。不能復(fù)制的原因和互斥鎖一樣。一旦讀寫鎖被使用，它的字段就會(huì)記錄它當(dāng)前的一些狀態(tài)。這個(gè)時(shí)候你去復(fù)制這把鎖，就會(huì)把它的狀態(tài)也給復(fù)制過來。但是，原來的鎖在釋放的時(shí)候，并不會(huì)修改你復(fù)制出來的這個(gè)讀寫鎖，這就會(huì)導(dǎo)致復(fù)制出來的讀寫鎖的狀態(tài)不對(duì)，可能永遠(yuǎn)無(wú)法釋放鎖。

不可重入

不可重入的原因是，獲得鎖之后，還沒釋放鎖，又申請(qǐng)鎖，這樣有可能造成死鎖。比如readerA獲取到了讀鎖，writerB等待readerA釋放鎖，reader還沒釋放鎖又申請(qǐng)了一把鎖，但是這把鎖申請(qǐng)不成功，他需要等待writerB。這就形成了一個(gè)循環(huán)等待的死鎖。

加鎖和釋放鎖一定要成對(duì)出現(xiàn)，不能忘記釋放鎖，也不能解鎖一個(gè)未加鎖的鎖。

實(shí)戰(zhàn)一下

Go中的map是不支持并發(fā)寫的，我們可以利用讀寫鎖RWMutex來實(shí)現(xiàn)并發(fā)安全的map。在讀多寫少的情況下，使用RWMutex要比Mutex性能高。

package

main

import

(

"fmt"

"math/rand"

"sync"

"time"

type

ConcurrentMap

struct

{

m

sync.RWMutex

items

map[string]interface{}

func

(c

*ConcurrentMap)

Add(key

string,

value

interface{})

{

c.m.Lock()

defer

c.m.Unlock()

c.items[key]

=

value

func

(c

*ConcurrentMap)

Remove(key

string)

{

c.m.Lock()

defer

c.m.Unlock()

delete(c