Go RWMutex 源碼學習

概述

上一篇文章咱們看go了互斥鎖的具體實現。可是若是業務邏輯是讀多寫少，若是每次讀寫都使用互斥鎖那麼整個效率就會變得很低。其實若是隻是讀的話並不須要互斥鎖來鎖住數據。只有寫操做的時候須要互斥鎖，可是若是有人讀那麼寫操做也應該被鎖住。
在Go語言中提供了讀寫鎖：RWMutex，而且提供了4個方法 讀鎖、讀解鎖、寫鎖、寫解鎖。其中讀鎖不是互斥，可是讀鎖和寫鎖是互斥的。簡單來講是能夠有多個讀同時加鎖，可是一旦有人想要獲取寫鎖則會被阻塞。

簡單使用

咱們能夠看到讀鎖能夠獲取多個，可是讀鎖還剩下一個的時候想要獲取寫鎖則會被阻塞。等待3秒以後讀鎖被所有解開以後，會喚醒以前阻塞的寫鎖。別忘記最後須要解開寫鎖。還有一個比較常見的問題是，若是給沒有讀鎖或者寫鎖的狀況下解鎖被拋出錯誤。

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    rw := sync.RWMutex{}
    rw.RLock()   
    rw.RLock()   
    rw.RLock()   
    rw.RUnlock() 
    rw.RUnlock() 
    go func() {
        time.Sleep(time.Second * 3)
        rw.RUnlock()
    }()
    fmt.Println("lock")
    rw.Lock() 
    rw.Unlock() 
    fmt.Println("unlock")
}

源碼分析

RWMutex實體

type RWMutex struct {
    // 內部鎖
    w           Mutex  
    // 寫信號量
    writerSem   uint32 
    // 讀信號量
    readerSem   uint32 
    // 準備讀的goroutine的數量
    readerCount int32  
    // 離開讀的goroutine的數量
    readerWait  int32  
}
// 讀寫鎖最大數量 1073741824
const rwmutexMaxReaders = 1 << 30

RLock、RUnlock、Lock、

// 加讀鎖
func (rw *RWMutex) RLock() {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    // 使用原子操做增長讀的數量操做readerCount + 1
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // 若是小於0 則掛起goroutine等待readerSem
        runtime_Semacquire(&rw.readerSem)
    }
    if race.Enabled {
        race.Enable()
        race.Acquire(unsafe.Pointer(&rw.readerSem))
    }
}

// 解讀鎖
func (rw *RWMutex) RUnlock() {
    if race.Enabled {
        _ = rw.w.state
        race.ReleaseMerge(unsafe.Pointer(&rw.writerSem))
        race.Disable()
    }
    // 設置readerCount - 1 記錄返回結果r
    if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
        // 若是r < 0 則報錯 若是沒有加鎖的狀況下解鎖則會報錯
        if r+1 == 0 || r+1 == -rwmutexMaxReaders {
            race.Enable()
            throw("sync: RUnlock of unlocked RWMutex")
        }
        // readerWait數量-1 
        if atomic.AddInt32(&rw.readerWait, -1) == 0 {
            // 若是度等待等於0，則恢復寫信號量的goroutine
            runtime_Semrelease(&rw.writerSem, false)
        }
    }
    if race.Enabled {
        race.Enable()
    }
}

// 寫鎖
func (rw *RWMutex) Lock() {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    // 第一步，先利用互斥鎖 加鎖
    rw.w.Lock()
    // 設置readerCount -1073741824 
    // 記錄返回值r  r再加上1073741824 獲取讀鎖的數量
    // 好比readerCount = 1   r = (1-1073741824) + 1073741824 = 1 
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    // 判斷讀等待是否不等於0 若是不爲0則阻塞等待
    if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
        runtime_Semacquire(&rw.writerSem)
    }
    if race.Enabled {
        race.Enable()
        race.Acquire(unsafe.Pointer(&rw.readerSem))
        race.Acquire(unsafe.Pointer(&rw.writerSem))
    }
}

func (rw *RWMutex) Unlock() {
    if race.Enabled {
        _ = rw.w.state
        race.Release(unsafe.Pointer(&rw.readerSem))
        race.Release(unsafe.Pointer(&rw.writerSem))
        race.Disable()
    }

    // 記錄並設置readerCount，使得readerCount爲正數 
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {
        race.Enable()
        // 未加鎖
        throw("sync: Unlock of unlocked RWMutex")
    }
    // 循環喚醒等待的讀型號量的goroutine
    for i := 0; i < int(r); i++ {
        runtime_Semrelease(&rw.readerSem, false)
    }
    // Allow other writers to proceed.
    rw.w.Unlock()
    if race.Enabled {
        race.Enable()
    }
}

用例子來分析源碼

仍是用上面的簡單的例子看仔細看RWMutex中屬性的變化。
下面代碼能夠看到主要的兩個屬性readerCount和readerWait兩個屬性的變化。
用最簡單的總結一下：

• RLock： readerCount + 1，獲得的結果readerCount < 0 此時有寫鎖，則掛起線程。
• RUnlock：readerCount - 1，獲得結果readerCount < 0 則readerWait--， 若是readerWait（讀等待）= 0 則喚醒寫操做阻塞。
• Lock：readerCount - rwmutexMaxReaders（1073741824），獲得結果再加上rwmutexMaxReaders獲取等待數量存入readerWait中。若是讀鎖不爲0 則阻塞寫鎖。
• Unlock：readerCount + rwmutexMaxReaders（1073741824），獲得等待的讀鎖個數而後循環喚醒全部讀等待的goroutine。

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    rw := sync.RWMutex{}
    rw.RLock()   // readerCount = 1; readerWait = 0
    rw.RLock()   // readerCount = 2; readerWait = 0
    rw.RLock()   // readerCount = 3; readerWait = 0
    rw.RUnlock() // readerCount = 2; readerWait = 0
    rw.RUnlock() // readerCount = 1; readerWait = 0
    go func() {
        time.Sleep(time.Second * 3)
        rw.RUnlock()
    }()
    fmt.Println("lock")
    rw.Lock()   // readerCount = -1073741824; readerWait = 0
    rw.Unlock() // readerCount = 0; readerWait = 0
    fmt.Println("unlock")
}

總結

互斥鎖能夠避免多線程中對同一個資源操做形成的問題，可是若是這個資源大部分狀況下是讀取少部分是寫操做，則推薦使用讀寫鎖來替換互斥鎖。能夠極大的提供效率，可是讀寫鎖的操做比互斥鎖多，有鎖和寫鎖兩種。若是操做不當很容易形成死鎖。因此加鎖和解鎖必需要保證是成對出現，而且考慮若是報錯的狀況下如何保證解鎖操做。