Go并發之RWMutex源碼分析

這篇文章主要介紹“Go并發之RWMutex源碼分析”的相關知識，小編通過實際案例向大家展示操作過程，操作方法簡單快捷，實用性強，希望這篇“Go并發之RWMutex源碼分析”文章能幫助大家解決問題。

RWMutex是一個支持并行讀串行寫的讀寫鎖。RWMutex具有寫操作優先的特點，寫操作發生時，僅允許正在執行的讀操作執行，后續的讀操作都會被阻塞。

使用場景

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

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

// 基于RWMutex的實現
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()
}

// 多個協程并行讀取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實現
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()
}

// 多個協程并行讀取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是一個寫操作優先的讀寫鎖，如下圖所示：

• 寫操作C發生時，讀操作A和讀操作B正在執行，因此寫操作C被掛起；

• 當讀操作D發生時，由于存在寫操作C等待鎖，所以讀操作D被掛起；

• 讀操作A和讀操作B執行完成，由于沒有讀操作和寫操作正在執行，寫操作C被喚醒執行；

• 當讀操作E發生時，由于寫操作C正在執行，所以讀操作E被掛起；

• 當寫操作C執行完成后，讀操作D和讀操作E被喚醒；

RWMutex結構體

RWMutex由如下變量組成：

• rwmutexMaxReaders：表示RWMutex能接受的最大讀協程數量，超過rwmutexMaxReaders后會發生panic；

• w：Mutex互斥鎖，用于實現寫操作之間的互斥

• writerSem：寫操作操作信號量；當存在讀操作時，寫操作會被掛起；讀操作全部完成后，通過writerSem信號量喚醒寫操作；

• readerSem：讀操作信號量；當存在寫操作時，讀操作會被掛起；寫操作完成后，通過readerSem信號量喚醒讀操作；

• readerCount：正在執行中的讀操作數量；當不存在寫操作時從0開始計數，為正數；當存在寫操作時從負的rwmutexMaxReaders開始計數，為負數；

• readerWait：寫操作等待讀操作的數量；當執行Lock()方法時，如果當前存在讀操作，會將讀操作的數量記錄在readerWait中，并掛起寫操作；讀操作執行完成后，會更新readerWait，當readerWait為0時，喚醒寫操作；

const rwmutexMaxReaders = 1 << 30

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

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

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

Lock()方法

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

• 獲取w互斥鎖，保證同一時刻只有一個寫操作執行；

• 將readerCount更新為負數，使后續發生的讀操作被阻塞；

• 如果當前存在活躍的讀操作r != 0，寫操作進入阻塞狀態runtime_SemacquireMutex；

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

Unlock()方法

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

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

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

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

func (rw *RWMutex) Unlock() {
    // 將readerCount更新為正數，從0開始計數
    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，讀操作進入阻塞狀態；

func (rw *RWMutex) RLock() {
    // 原子更新readerCount+1
    // 1. readerCount+1為負數時，表示當前存在寫操作；讀操作需要等待寫操作執行完，才能被執行
    // 2. readerCount+1不為負數時，表示當前不存在寫操作，讀操作可以執行
    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時，表示阻塞寫操作的所有讀操作都執行完了，喚醒寫操作；

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時，表示導致寫操作阻塞的所有讀操作都執行完，將寫操作喚醒
    if atomic.AddInt32(&rw.readerWait, -1) == 0 {
        // 喚醒讀操作
        runtime_Semrelease(&rw.writerSem, false, 1)
    }
}

關于“Go并發之RWMutex源碼分析”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識，可以關注億速云行業資訊頻道，小編每天都會為大家更新不同的知識點。