Golang 之 RWMutex 源码解读

type RWMutex struct {
    // 写操作锁，通过它实现写写操作互斥
    w           Mutex
    // 写锁信号量
    writerSem   uint32
    // 读锁信号量
    readerSem   uint32
    // 当前读操作的数量，包含所有已经获取到读锁或者被写操作阻塞的等待获取读锁的读操作数量
    readerCount int32
    // 获取写锁需要等待读锁释放的数量
    readerWait  int32
}
复制

func (rw *RWMutex) Lock() {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    // 首先竞争获取到读锁
    // 这里实现的是写写互斥
    rw.w.Lock()
    // 接下来是读写互斥
    // 首先需要通知读锁，有一个写操作被阻塞了
    // 这里正是设计的巧妙之处
    // 如果这里赋值成功，那么后续读锁加锁的时候，都不可能使 readerCount > 0
    // 当然以上是基于读并发数量不可能 > rwmutexMaxReaders 这个前提
    // 当读加锁的时候发现 readerCount < 0 了，就可以知道当前有被阻塞的写操作
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    // 等待读锁全部释放，两个条件，满足其一即可：
    // 1. r = 0
    // 2. atomic.AddInt32(&rw.readerWait, r) = 0
    // 否则需要阻塞等待写信号量
    if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
        // 和 RUnlock() 释放写锁信号量相呼应
        runtime_SemacquireMutex(&rw.writerSem, false, 0)
    }
    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.Disable()
    }


    // 和加锁时减 rwmutexMaxReaders 相呼应，告诉读协程没有活跃中的写操作了
    // 这之后 readerCount 又能恢复正常(> 0)，代表当前读操作的数量
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    // 异常情况
    if r >= rwmutexMaxReaders {
        race.Enable()
        throw("sync: Unlock of unlocked RWMutex")
    }
    // 唤醒读操作，与下文 RLock() 读操作等待信号量相呼应
    for i := 0; i < int(r); i++ {
        runtime_Semrelease(&rw.readerSem, false, 0)
    }
    // 释放写锁
    rw.w.Unlock()
    if race.Enabled {
        race.Enable()
    }
}
复制

func (rw *RWMutex) RLock() {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    // 如果 atomic.AddInt32(&rw.readerCount, 1) < 0，代表写锁正在被占用，需要等待写锁释放
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // 阻塞等待写锁释放读信号量，和上文 Unlock() 释放读锁信号量相呼应
        runtime_SemacquireMutex(&rw.readerSem, false, 0)
    }
    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
    // 如果此时 readerCount < 0，代表有写操作被阻塞，需要走 slow-path，择机唤醒写操作
    if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
        // Outlined slow-path to allow the fast-path to be inlined
        rw.rUnlockSlow(r)
    }
    if race.Enabled {
        race.Enable()
    }
}
复制

// r -> rw.readerCount，此时是读锁解锁减去 1 之后的值
func (rw *RWMutex) rUnlockSlow(r int32) {
    // 如果 r+1 = 0，则 r = -1，代表解锁了一个未加锁的锁
    // 如果 r+1 = -rwmutexMaxReaders，还记得写锁加锁的操作嘛，这里代表解锁了一个写锁
    // 这两种情况都是非法的
    if r+1 == 0 || r+1 == -rwmutexMaxReaders {
        race.Enable()
        throw("sync: RUnlock of unlocked RWMutex")
    }
    // 写操作需要等待读操作释放的数量减 1
    // 如果此时值变成了 0，代表前面的读操作全部释放了
    if atomic.AddInt32(&rw.readerWait, -1) == 0 {
        // 唤醒一个写锁操作，和 Lock() 等待写锁信号量相呼应
        runtime_Semrelease(&rw.writerSem, false, 1)
    }
}
复制

func (rw *RWMutex) RLocker() Locker {
    return (*rlocker)(rw)
}


type rlocker RWMutex


func (r *rlocker) Lock()   { (*RWMutex)(r).RLock() }
func (r *rlocker) Unlock() { (*RWMutex)(r).RUnlock() }
复制

// 回源函数
type Load func(string) interface{}


type Cache interface {
     Get 从缓存中取值，如果缓存中没有，那么通过 load 方法回源并加载进缓存
    Get(string, Load) interface{}


    // Put 往缓存中写数据
    Put(key, value string)
}
复制

// 利用 RWMutex 实现一个缓存
// 缓存是一个典型的读多写少的系统，这与读写锁的使用场景不谋而合
type MCache struct {
    cacheMap map[string]interface{}
    sync.RWMutex
}


// Get 读，按需加载
func (mc *MCache) Get(key string, load func(string) interface{}) (value interface{}) {
    mc.RLock()
    if val, exist := mc.cacheMap[key]; exist {
        value = val
        mc.RUnlock()
        return value
    }
    mc.Lock()
    defer mc.Unlock()
    // 双重检查
    if val, exist := mc.cacheMap[key]; exist {
        value = val
        return
    }
    // 查询数据源并写到缓存中
    value = load(key)
    mc.cacheMap[key] = value
    return
}


// Put 写，需要添加写锁
func (mc *MCache) Put(key, value interface{}) {
    mc.Lock()
    mc.cacheMap[key] = value
    mc.Unlock()
}
复制