使用Golang的singleflight防止缓存击穿的方法

import "golang.org/x/sync/singleflight"

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

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

func main() {

  var singleSetCache singleflight.Group

  getAndSetCache:=func (requestID int,cacheKey string) (string, error) {

  log.Printf("request %v start to get and set cache...",requestID)

  value,_, _ :=singleSetCache.Do(cacheKey, func() (ret interface{}, err error) {//do的入参key，可以直接使用缓存的key，这样同一个缓存，只有一个协程会去读DB

    log.Printf("request %v is setting cache...",requestID)

     time.Sleep(3*time._Second_)

     log.Printf("request %v set cache success!",requestID)

    return "VALUE",nil

   })

  return value.(string),nil

  }

  cacheKey:="cacheKey"

  for i:=1;i<10;i++{//模拟多个协程同时请求

  go func(requestID int) {

     value,_:=getAndSetCache(requestID,cacheKey)

     log.Printf("request %v get value: %v",requestID,value)

   }(i)

  }

  time.Sleep(20*time._Second_)
}

type Group struct {

  mu sync.Mutex   

  m map[string]*call //保存key对应的函数执行过程和结果的变量。

}

type call struct {

  wg sync.WaitGroup //用WaitGroup实现只有一个协程执行函数

  val interface{} //函数执行结果

  err error

  forgotten bool

  dups int //含义是duplications，即同时执行同一个key的协程数量

  chans []chan<- Result
}

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

  g.mu.Lock()//写Group的m字段时，加锁保证写安全。

  if g.m == nil {

  g.m = make(map[string]*call)

  }

if c, ok := g.m[key]; ok {//如果key已经存在，说明已经有协程在执行，则dups++，并等待其执行完毕后，返回其执行结果，执行结果保存在对应的call的val字段里

   c.dups++

   g.mu.Unlock()

   c.wg.Wait()

 return c.val, c.err, true

  }

//如果key不存在，则新建一个call，并使用WaitGroup来阻塞其他协程，同时在m字段里写入key和对应的call

c := new(call)

  c.wg.Add(1)

  g.m[key] = c

  g.mu.Unlock()

  g.doCall(c, key, fn)//第一个进来的协程来执行这个函数

return c.val, c.err, c.dups > 0

}

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {

  c.val, c.err = fn()//执行被包装的函数

  c.wg.Done()//执行完毕后，就可以通知其他协程可以拿结果了

  g.mu.Lock()

if !c.forgotten {//其实这里是为了保证执行完毕之后，对应的key被删除，Group有一个方法Forget（key string），可以用来主动删除key，这里是判断那个方法是否被调用过，被调用过则字段forgotten会置为true，如果没有被调用过，则在这里把key删除。

  delete(g.m, key)

  }

  for _, ch := range c.chans {//将执行结果发送到channel里，这里是给DoChan方法使用的

  ch <- Result{c.val, c.err, c.dups > 0}

  }

  g.mu.Unlock()

}

func main() {

  var singleSetCache singleflight.Group

  getAndSetCache:=func (requestID int,cacheKey string) (string, error) {

  log.Printf("request %v start to get and set cache...",requestID)

  retChan:=singleSetCache.DoChan(cacheKey, func() (ret interface{}, err error) {

    log.Printf("request %v is setting cache...",requestID)

    time.Sleep(3*time._Second_)

    log.Printf("request %v set cache success!",requestID)

    return "VALUE",nil

   })

  var ret singleflight.Result

  timeout := time.After(5 * time._Second_)

  select {//加入了超时机制

    case <-timeout:

      log.Printf("time out!")

      return "",errors.New("time out")

    case ret =<- retChan://从chan中取出结果

      return ret.Val.(string),ret.Err

   }

  return "",nil

  }

  cacheKey:="cacheKey"
  
  for i:=1;i<10;i++{

  go func(requestID int) {

     value,_:=getAndSetCache(requestID,cacheKey)

     log.Printf("request %v get value: %v",requestID,value)

   }(i)

  }

  time.Sleep(20*time._Second_)

}

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

  ch := make(chan Result, 1)

  g.mu.Lock()

  if g.m == nil {

  g.m = make(map[string]*call)

  }

  if c, ok := g.m[key]; ok {

   c.dups++

c.chans = append(c.chans, ch)//可以看到，每个等待的协程，都有一个结果channel。从之前的g.doCall里也可以看到，每个channel都给塞了结果。为什么不所有协程共用一个channel？因为那样就得在channel里塞至少与协程数量一样的结果数量，但是你却无法保证用户一个协程只读取一次。

   g.mu.Unlock()

   return ch

  }

  c := &call{chans: []chan<- Result{ch}}

  c.wg.Add(1)

  g.m[key] = c

  g.mu.Unlock()

  go g.doCall(c, key, fn)

  return ch
}