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")
}

总结

互斥锁能够避免多线程中对同一个资源操做形成的问题，可是若是这个资源大部分状况下是读取少部分是写操做，则推荐使用读写锁来替换互斥锁。能够极大的提供效率，可是读写锁的操做比互斥锁多，有锁和写锁两种。若是操做不当很容易形成死锁。因此加锁和解锁必需要保证是成对出现，而且考虑若是报错的状况下如何保证解锁操做。