上篇 介绍了通过Redis的分布式锁解决对缓存数据的存取过程中，如果高并发而造成的重复获取数据、甚至造成数据的不一致性的问题。Redis只是一种解决方案，并且适合多节点多副本的场景。

如果单节点或者没有Redis的情况下，我们能否实现呢。答案是当然可以的，只要不断探寻，总能解决的。本篇就介绍通过Go的传统锁，select和channel来实现这样的高并发数据缓存的存取逻辑。

为了便于读者看本篇内容和代码时，更易于理解，本篇还引用上篇的流程草图，但具体流程、场景不再细说，只是本篇不采用redis，而是把数据缓存到内存中。

贴代码前，先看2个知识点，有助于理解本篇主题的代码，Go的读写锁RWMutex和select结合channel的多路复用的用法。

读写锁RWMutex

名词解释

• 读锁，写锁是一个名词
• 读锁定RLock()，写锁定Lock()是一个动作、操作、函数调用
• 读解锁RUnlock()，写解锁Unlock()是一个动作、操作、函数调用

RWMutex是一个专门提供读或写的锁，这个锁可以被很多读锁锁定，或被某个写锁锁定，有如下特性，也就保证了操作数据的原子性。

• 读锁之间不互斥，读锁与写锁，写锁与写锁之间都会互
• 适合多读少写的场景，因为读锁之间不互斥
• 当读锁锁定的时候，且此时有写锁锁定调用，会阻止任何一个协程来获取这个锁，包括读锁
• 不管是读锁还是写锁，如果之前没有锁定，此时调用解锁会Panic runtime error
• 不独享某个线程，可以某个线程锁定，另外某个线程解锁

这些特性我们通过代码只验证这一条“ 当读锁锁定的时候，且此时有写锁锁定调用，会阻止任何一个协程来获取这个锁，包括读锁 ”，其他比较容易理解和自行验证。

 var rwmu sync.RWMutex

func RWMuFoo() {
  go func() {
    time.Sleep(time.Millisecond * 100)
    for i := 0; i < 5; i++ {
      go func(a int) {
        if a/2 == 0 {
          rwmu.Lock()
          fmt.Println("wlock")
          rwmu.Unlock()
        } else {
          time.Sleep(time.Millisecond * 100)
          rwmu.RLock()
          fmt.Println("rlock")
          rwmu.RUnlock()
        }
      }(i)
    }
  }()

  rwmu.RLock()
  fmt.Println("start to rlock")
  time.Sleep(time.Second)
  fmt.Println("end to rlock")
  rwmu.RUnlock()
}  

• 第1行声明了一个读写锁的变量rwmu
• 第22-26行，演示进行读锁定，并且停留一秒钟，才进行读解锁。停留一秒钟是为了让其他协程有机会进行读锁定或写锁定。
• 第4-20行，启动协程来对rwmu继续读锁定和写锁定。由于协程调度的机制并不能保证代码写的那样的顺序执行，所以为了演示该效果，代码中有2处停留
• 第5行停留100毫秒是为了让第22-26行代码先执行。
• 第13行停留100毫秒是为了让第9-11行的写锁先锁定。

所以以上代码可以验证“ 当读锁锁定的时候，且此时有写锁锁定调用，会阻止任何一个协程来获取这个锁，包括读锁 ”。现在运行代码

   c := make(chan int)
  select {
  case c <- 1:
    fmt.Println("c<-1")
  default:
    fmt.Println("default")
  }  

是符合预期的。

select, channel

   c := make(chan int)
  select {
  case c <- 1:
    fmt.Println("c<-1")
  default:
    fmt.Println("default")
  }  

select语句可以让一个协程等待多个通信操作。如果没有default语句时，select会一直阻塞，知道某个case可以执行，如果同时有多个case均可以执行，系统会随机选择一个来执行。

channel的概念不多讲，有兴趣的可以看看官方文档。

数据缓存的高并发存取

通过对RWMutex, select, channel的介绍，我们就可以开发类似上篇的逻辑代码了。

代码片段1

 // NBLock likes redis's set nx
type NBLock struct {
  locked chan bool
}

func (l NBLock) lock() bool {
  select {
  case l.locked <- true:
    fmt.Println("locked")
    return true
  default:
    return false
  }
}

func (l NBLock) unlock() {
  <-l.locked
}

var nblock = NBLock{locked: make(chan bool, 1)}
var result string

var rwmu sync.RWMutex

func HandleReq() {
  // rwmu.RLock()
  reslen := len(result)
  // rwmu.RUnlock()
  if reslen > 0 {
    fmt.Println("get global var: ok")
    return
  }

  if nblock.lock() {
    // rwmu.Lock()
    result = fetch()
    fmt.Println("fetch: ok")
    // rwmu.Unlock()

    nblock.unlock()
  } else {
    for {
      // rwmu.RLock()
      reslen1 := len(result)
      // rwmu.RUnlock()
      if reslen1 > 0 {
        fmt.Println("waiting: ok ")
        return
      }
    }
  }
}  

• 第1-17行，自定义了一个非阻塞的锁，并且只能锁定成功一次，在没有解锁之前，不能再次进行解锁。非常类似Redis的分布式锁。
• 第25-52行的逻辑比较类似上篇的逻辑。即高并发请求进来，会锁定第一次的数据获取，其他请求则等待或从缓存获取。

现在运行并发10次运行代码

   for i := 0; i < 10; i++ {
    if i > 4 {
      time.Sleep(time.Millisecond * 100)
    }
    go igo.ONReqWithRWMu()
  }  

运行结果如下

 locked
waiting: ok 
get global var: ok
waiting: ok 
waiting: ok 
fetch: ok
waiting: ok 
get global var: ok
get global var: ok
get global var: ok
get global var: ok  

结果是符合我们预期的，lock了一次，获取数据fetch了一次，剩下的有等待的，也有直接从变量获取的。

咦，有没搞错，怎么没有用到读写锁呢RWMutex（其实先注释掉了）。这段代码里为什么要用读写锁，现在我们添加 -race 竟态检查运行下代码，就知道了。

 Write at 0x000001667a90 by goroutine 8:
  interview-go/igo.ONReqWithRWMu()
      /.../onenode_waitresult.go:36 +0x125

Previous read at 0x000001667a90 by goroutine 10:
  interview-go/igo.ONReqWithRWMu()
      /.../onenode_waitresult.go:44 +0x87
==================  

运行结果告诉我们

• 协程8在代码片段1的36行处result = fetch()对result赋值。
• 协程10在代码片段1的44行处reslen1 := len (result)对result读取长度。
• 这2个协程发生了对变量的竞争，如果不处理，很可能会造成数据的不一致或错乱。

此时读写锁就发挥出作用了，现在将代码片段1中的关于读写锁的注释去掉，即第26、28、35、38、43、45行去掉注释，然后运行代码，并加上-race进行竟态检查

 go test ./test/onenode_waitresult_test.go -v -count=1 -race  

 locked
fetch: ok
get global var: ok
get global var: ok
get global var: ok
get global var: ok
waiting: ok 
get global var: ok
get global var: ok
get global var: ok
get global var: ok  

运行结果也符合预期，且没有了对result的竞争。

本篇内容基本就到这里。

总结

• 本篇内容和上篇内容还是有区别，并不能很好地通过设置timeout来强制等待的协程返回内容。
• 可以通过在fetch的方法里面进行判断timeout或出错后，一定要写解锁。否则其他协程就会一直处于阻塞状态。
• 本篇代码更多是一种实现的思想，也更好地介绍了读写锁RWMutex和自己通过select, channel来实现一个非阻塞的锁，且没解锁前只能成功锁定一次。

篇篇更精彩，章章有深度

感谢大家转发、关注和一起讨论学习