Go并發(fā)之RWMutex的源碼解析詳解

更新時間：2023年03月15日 09:49:22 作者：Shine4YG

RWMutex是一個支持并行讀串行寫的讀寫鎖。RWMutex具有寫操作優(yōu)先的特點，寫操作發(fā)生時，僅允許正在執(zhí)行的讀操作執(zhí)行，后續(xù)的讀操作都會被阻塞。本文就來從源碼解析一下RWMutex的使用

使用場景

RWMutex常用于大量并發(fā)讀，少量并發(fā)寫的場景；比如微服務配置更新、交易路由緩存等場景。相對于Mutex互斥鎖，RWMutex讀寫鎖具有更好的讀性能。

下面以 “多個協(xié)程并行讀取str變量，一個協(xié)程每100毫秒定時更新str變量” 場景為例，進行RWMutex讀寫鎖和Mutex互斥鎖的性能對比。

// 基于RWMutex的實現(xiàn)
var rwLock sync.RWMutex
var str1 = "hello"

func readWithRWLock() string {
    rwLock.RLock()
    defer rwLock.RUnlock()
    return str1
}

func writeWithRWLock() {
    rwLock.Lock()
    str1 = time.Now().Format("20060102150405")
    rwLock.Unlock()
}

// 多個協(xié)程并行讀取string變量，同時每100ms對string變量進行1次更新
func BenchmarkRWMutex(b *testing.B) {
    ticker := time.NewTicker(100 * time.Millisecond)
    go func() {
        for range ticker.C {
            writeWithRWLock()
        }
    }()
    b.ResetTimer()
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            readWithRWLock()
        }
    })
}
// 基于Mutex實現(xiàn)
var lock sync.Mutex
var str2 = "hello"

func readWithMutex() string {
    lock.Lock()
    defer lock.Unlock()
    return str2
}

func writeWithMutex() {
    lock.Lock()
    str2 = time.Now().Format("20060102150405")
    lock.Unlock()
}

// 多個協(xié)程并行讀取string變量，同時每100ms對string變量進行1次更新
func BenchmarkMutex(b *testing.B) {
    ticker := time.NewTicker(100 * time.Millisecond)
    go func() {
        for range ticker.C {
            writeWithMutex()
        }
    }()
    b.ResetTimer()
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            readWithMutex()
        }
    })
}

RWMutex讀寫鎖和Mutex互斥鎖的性能對比，結果如下：

# go test 結果
go test -bench . -benchtime=10s
BenchmarkRWMutex-8 227611413 49.5 ns/op
BenchmarkMutex-8 135363408 87.8 ns/op
PASS
ok demo 37.800s

源碼解析

RWMutex是一個寫操作優(yōu)先的讀寫鎖，如下圖所示：

寫操作C發(fā)生時，讀操作A和讀操作B正在執(zhí)行，因此寫操作C被掛起；
當讀操作D發(fā)生時，由于存在寫操作C等待鎖，所以讀操作D被掛起；
讀操作A和讀操作B執(zhí)行完成，由于沒有讀操作和寫操作正在執(zhí)行，寫操作C被喚醒執(zhí)行；
當讀操作E發(fā)生時，由于寫操作C正在執(zhí)行，所以讀操作E被掛起；
當寫操作C執(zhí)行完成后，讀操作D和讀操作E被喚醒；

RWMutex結構體

RWMutex由如下變量組成：

rwmutexMaxReaders：表示RWMutex能接受的最大讀協(xié)程數(shù)量，超過rwmutexMaxReaders后會發(fā)生panic；
w：Mutex互斥鎖，用于實現(xiàn)寫操作之間的互斥
writerSem：寫操作操作信號量；當存在讀操作時，寫操作會被掛起；讀操作全部完成后，通過writerSem信號量喚醒寫操作；
readerSem：讀操作信號量；當存在寫操作時，讀操作會被掛起；寫操作完成后，通過readerSem信號量喚醒讀操作；
readerCount：正在執(zhí)行中的讀操作數(shù)量；當不存在寫操作時從0開始計數(shù)，為正數(shù)；當存在寫操作時從負的rwmutexMaxReaders開始計數(shù)，為負數(shù)；
readerWait：寫操作等待讀操作的數(shù)量；當執(zhí)行Lock()方法時，如果當前存在讀操作，會將讀操作的數(shù)量記錄在readerWait中，并掛起寫操作；讀操作執(zhí)行完成后，會更新readerWait，當readerWait為0時，喚醒寫操作；

const rwmutexMaxReaders = 1 << 30

type RWMutex struct {
    w           Mutex  // Mutex互斥鎖，用于實現(xiàn)寫操作之間的互斥

    writerSem   uint32 // 寫操作信號量，用于讀操作喚醒寫操作
    readerSem   uint32 // 讀操作信號量，用于寫操作喚醒讀操作

    readerCount int32  // 讀操作的數(shù)量，不存在寫操作時從0開始計數(shù)，存在寫操作時從-rwmutexMaxReaders開始計數(shù)
    readerWait  int32  // 寫操作等待讀操作的數(shù)量
}

Lock()方法

Lock方法用于寫操作獲取鎖，其操作如下：

獲取w互斥鎖，保證同一時刻只有一個寫操作執(zhí)行；
將readerCount更新為負數(shù)，使后續(xù)發(fā)生的讀操作被阻塞；
如果當前存在活躍的讀操作r != 0，寫操作進入阻塞狀態(tài)runtime_SemacquireMutex；

func (rw *RWMutex) Lock() {
    // 寫操作之間通過w互斥鎖實現(xiàn)互斥
    rw.w.Lock()
    // 1.將readerCount更新為負值，表示當前有寫操作；當readerCount為負數(shù)時，新的讀操作會被掛起
    // 2.r表示當前正在執(zhí)行的讀操作數(shù)量
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    // r != 0表示當前存在正在執(zhí)行的讀操作；寫操作需要等待所有讀操作執(zhí)行完，才能被執(zhí)行；
    if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
        // 將寫操作掛起
        runtime_SemacquireMutex(&rw.writerSem, false, 0)
    }
}

Unlock()方法

Unlock方法用于寫操作釋放鎖，其操作如下：

將readerCount更新為正數(shù)，表示當前不存在活躍的寫操作；

如果更新后的readerCount大于0，表示當前寫操作阻塞了readerCount個讀操作，需要將所有被阻塞的讀操作都喚醒；

將w互斥鎖釋放，允許其他寫操作執(zhí)行；

func (rw *RWMutex) Unlock() {
    // 將readerCount更新為正數(shù)，從0開始計數(shù)
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {
        throw("sync: Unlock of unlocked RWMutex")
    }
    // 喚醒所有等待寫操作的讀操作 
    for i := 0; i < int(r); i++ {
        runtime_Semrelease(&rw.readerSem, false, 0)
    }
    // 釋放w互斥鎖，允許其他寫操作進入
    rw.w.Unlock()
}

RLock()方法

RLock方法用于讀操作獲取鎖，其操作如下：

原子更新readerCount+1；
如果當前存在寫操作atomic.AddInt32(&rw.readerCount, 1) < 0，讀操作進入阻塞狀態(tài)；

func (rw *RWMutex) RLock() {
    // 原子更新readerCount+1
    // 1. readerCount+1為負數(shù)時，表示當前存在寫操作；讀操作需要等待寫操作執(zhí)行完，才能被執(zhí)行
    // 2. readerCount+1不為負數(shù)時，表示當前不存在寫操作，讀操作可以執(zhí)行
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // 將讀操作掛起
        runtime_SemacquireMutex(&rw.readerSem, false, 0)
    }
}

RUnlock()方法

RUnlock方法用于讀操作釋放鎖，其操作如下：

原子更新readerCount-1；

如果當前讀操作阻塞了寫操作atomic.AddInt32(&rw.readerCount, -1)<0，原子更新readerWait-1；

當readerWait為0時，表示阻塞寫操作的所有讀操作都執(zhí)行完了，喚醒寫操作；

func (rw *RWMutex) RUnlock() {
    // 原子更新readerCount-1
    // 當readerCount-1為負時，表示當前讀操作阻塞了寫操作，需要進行readerWait的更新
    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 {
        throw("sync: RUnlock of unlocked RWMutex")
    }
    // 原子操作readerWait-1
    // 當readerWait-1為0時，表示導致寫操作阻塞的所有讀操作都執(zhí)行完，將寫操作喚醒
    if atomic.AddInt32(&rw.readerWait, -1) == 0 {
        // 喚醒讀操作
        runtime_Semrelease(&rw.writerSem, false, 1)
    }
}

到此這篇關于Go并發(fā)之RWMutex的源碼解析詳解的文章就介紹到這了,更多相關Go RWMutex內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: