使用Golang的singleflight防止缓存击穿
2021-04-18 14:54:49    245    0    0
admin

背景

在使用缓存时,容易发生缓存击穿。

缓存击穿:一个存在的key,在缓存过期的瞬间,同时有大量的请求过来,造成所有请求都去读数据库,这些请求都会击穿到数据库,造成瞬时请求量大、压力骤增。

singleflight

介绍
import "golang.org/x/sync/singleflight"
singleflight类的使用方法:
新建一个singleflight.Group,使用其方法Do或者DoChan来包装方法,被包装的方法对于同一个key只会有一个协程执行,其他协程等待那个协程执行结束后,拿到同样的结果。

Group结构体

代表一类工作,同一个group中,同样的key同时只能被执行一次。

Do方法

  1. func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool)

key:同一个key,同时只有一个协程执行。
fn:被包装的函数。
v:返回值,即执行的结果。其他等待的协程都会拿到。
shared:表示是否有其他协程得到了这个结果v。

DoChan方法

  1. func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result

与Do方法一样,只是返回的是一个channel,执行结果会发送到channel中,其他等待的协程都可以从channel中拿到结果。

ref:https://godoc.org/golang.org/x/sync/singleflight

示例

使用Do方法来模拟,解决缓存击穿的问题

  1. func main() {
  2. var singleSetCache singleflight.Group
  3. getAndSetCache:=func (requestID int,cacheKey string) (string, error) {
  4. log.Printf("request %v start to get and set cache...",requestID)
  5. value,_, _ :=singleSetCache.Do(cacheKey, func() (ret interface{}, err error) {//do的入参key,可以直接使用缓存的key,这样同一个缓存,只有一个协程会去读DB
  6. log.Printf("request %v is setting cache...",requestID)
  7. time.Sleep(3*time.Second)
  8. log.Printf("request %v set cache success!",requestID)
  9. return "VALUE",nil
  10. })
  11. return value.(string),nil
  12. }
  13. cacheKey:="cacheKey"
  14. for i:=1;i<10;i++{//模拟多个协程同时请求
  15. go func(requestID int) {
  16. value,_:=getAndSetCache(requestID,cacheKey)
  17. log.Printf("request %v get value: %v",requestID,value)
  18. }(i)
  19. }
  20. time.Sleep(20*time.Second)
  21. }

输出:

  1. 2020/04/12 18:18:40 request 4 start to get and set cache...
  2. 2020/04/12 18:18:40 request 4 is setting cache...
  3. 2020/04/12 18:18:40 request 2 start to get and set cache...
  4. 2020/04/12 18:18:40 request 7 start to get and set cache...
  5. 2020/04/12 18:18:40 request 5 start to get and set cache...
  6. 2020/04/12 18:18:40 request 1 start to get and set cache...
  7. 2020/04/12 18:18:40 request 6 start to get and set cache...
  8. 2020/04/12 18:18:40 request 3 start to get and set cache...
  9. 2020/04/12 18:18:40 request 8 start to get and set cache...
  10. 2020/04/12 18:18:40 request 9 start to get and set cache...
  11. 2020/04/12 18:18:43 request 4 set cache success!
  12. 2020/04/12 18:18:43 request 4 get value: VALUE
  13. 2020/04/12 18:18:43 request 9 get value: VALUE
  14. 2020/04/12 18:18:43 request 6 get value: VALUE
  15. 2020/04/12 18:18:43 request 3 get value: VALUE
  16. 2020/04/12 18:18:43 request 8 get value: VALUE
  17. 2020/04/12 18:18:43 request 1 get value: VALUE
  18. 2020/04/12 18:18:43 request 5 get value: VALUE
  19. 2020/04/12 18:18:43 request 2 get value: VALUE
  20. 2020/04/12 18:18:43 request 7 get value: VALUE

可以看到确实只有一个协程执行了被包装的函数,并且其他协程都拿到了结果。

源码分析

看一下这个Do方法是怎么实现的。
首先看一下Group的结构:

  1. type Group struct {
  2. mu sync.Mutex
  3. m map[string]*call //保存key对应的函数执行过程和结果的变量。
  4. }

Group的结构非常简单,一个锁来保证并发安全,另一个map用来保存key对应的函数执行过程和结果的变量。

看下call的结构:

  1. type call struct {
  2. wg sync.WaitGroup //用WaitGroup实现只有一个协程执行函数
  3. val interface{} //函数执行结果
  4. err error
  5. forgotten bool
  6. dups int //含义是duplications,即同时执行同一个key的协程数量
  7. chans []chan<- Result
  8. }

看下Do方法

  1. func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
  2. g.mu.Lock()//写Group的m字段时,加锁保证写安全。
  3. if g.m == nil {
  4. g.m = make(map[string]*call)
  5. }
  6. if c, ok := g.m[key]; ok {//如果key已经存在,说明已经有协程在执行,则dups++,并等待其执行完毕后,返回其执行结果,执行结果保存在对应的call的val字段里
  7. c.dups++
  8. g.mu.Unlock()
  9. c.wg.Wait()
  10. return c.val, c.err, true
  11. }
  12. //如果key不存在,则新建一个call,并使用WaitGroup来阻塞其他协程,同时在m字段里写入key和对应的call
  13. c := new(call)
  14. c.wg.Add(1)
  15. g.m[key] = c
  16. g.mu.Unlock()
  17. g.doCall(c, key, fn)//第一个进来的协程来执行这个函数
  18. return c.val, c.err, c.dups > 0
  19. }

继续看下g.doCall里具体干了什么

  1. func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
  2. c.val, c.err = fn()//执行被包装的函数
  3. c.wg.Done()//执行完毕后,就可以通知其他协程可以拿结果了
  4. g.mu.Lock()
  5. if !c.forgotten {//其实这里是为了保证执行完毕之后,对应的key被删除,Group有一个方法Forget(key string),可以用来主动删除key,这里是判断那个方法是否被调用过,被调用过则字段forgotten会置为true,如果没有被调用过,则在这里把key删除。
  6. delete(g.m, key)
  7. }
  8. for _, ch := range c.chans {//将执行结果发送到channel里,这里是给DoChan方法使用的
  9. ch <- Result{c.val, c.err, c.dups > 0}
  10. }
  11. g.mu.Unlock()
  12. }

由此看来,其实现是非常简单的。不得不赞叹一百来行代码就实现了功能。

其他

顺便附上DoChan方法的使用示例:

  1. func main() {
  2. var singleSetCache singleflight.Group
  3. getAndSetCache:=func (requestID int,cacheKey string) (string, error) {
  4. log.Printf("request %v start to get and set cache...",requestID)
  5. retChan:=singleSetCache.DoChan(cacheKey, func() (ret interface{}, err error) {
  6. log.Printf("request %v is setting cache...",requestID)
  7. time.Sleep(3*time.Second)
  8. log.Printf("request %v set cache success!",requestID)
  9. return "VALUE",nil
  10. })
  11. var ret singleflight.Result
  12. timeout := time.After(5 * time.Second)
  13. select {//加入了超时机制
  14. case <-timeout:
  15. log.Printf("time out!")
  16. return "",errors.New("time out")
  17. case ret =<- retChan://从chan中取出结果
  18. return ret.Val.(string),ret.Err
  19. }
  20. return "",nil
  21. }
  22. cacheKey:="cacheKey"
  23. for i:=1;i<10;i++{
  24. go func(requestID int) {
  25. value,_:=getAndSetCache(requestID,cacheKey)
  26. log.Printf("request %v get value: %v",requestID,value)
  27. }(i)
  28. }
  29. time.Sleep(20*time.Second)
  30. }

看下DoChan的源码

  1. func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result {
  2. ch := make(chan Result, 1)
  3. g.mu.Lock()
  4. if g.m == nil {
  5. g.m = make(map[string]*call)
  6. }
  7. if c, ok := g.m[key]; ok {
  8. c.dups++
  9. c.chans = append(c.chans, ch)//可以看到,每个等待的协程,都有一个结果channel。从之前的g.doCall里也可以看到,每个channel都给塞了结果。为什么不所有协程共用一个channel?因为那样就得在channel里塞至少与协程数量一样的结果数量,但是你却无法保证用户一个协程只读取一次。
  10. g.mu.Unlock()
  11. return ch
  12. }
  13. c := &call{chans: []chan<- Result{ch}}
  14. c.wg.Add(1)
  15. g.m[key] = c
  16. g.mu.Unlock()
  17. go g.doCall(c, key, fn)
  18. return ch
  19. }

Prev: Go Mod 添加私有仓库

Next: 创建swap内存-华硕AX82U路由器官方原版固件

245
Table of content