如何使用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的入?yún)ey，可以直接使用緩存的key，這樣同一個(gè)緩存，只有一個(gè)協(xié)程會(huì)去讀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++{//模擬多個(gè)協(xié)程同時(shí)請(qǐng)求

  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對(duì)應(yīng)的函數(shù)執(zhí)行過(guò)程和結(jié)果的變量。

}

type call struct {

  wg sync.WaitGroup //用WaitGroup實(shí)現(xiàn)只有一個(gè)協(xié)程執(zhí)行函數(shù)

  val interface{} //函數(shù)執(zhí)行結(jié)果

  err error

  forgotten bool

  dups int //含義是duplications，即同時(shí)執(zhí)行同一個(gè)key的協(xié)程數(shù)量

  chans []chan<- Result
}

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

  g.mu.Lock()//寫Group的m字段時(shí)，加鎖保證寫安全。

  if g.m == nil {

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

  }

if c, ok := g.m[key]; ok {//如果key已經(jīng)存在，說(shuō)明已經(jīng)有協(xié)程在執(zhí)行，則dups++，并等待其執(zhí)行完畢后，返回其執(zhí)行結(jié)果，執(zhí)行結(jié)果保存在對(duì)應(yīng)的call的val字段里

   c.dups++

   g.mu.Unlock()

   c.wg.Wait()

 return c.val, c.err, true

  }

//如果key不存在，則新建一個(gè)call，并使用WaitGroup來(lái)阻塞其他協(xié)程，同時(shí)在m字段里寫入key和對(duì)應(yīng)的call

c := new(call)

  c.wg.Add(1)

  g.m[key] = c

  g.mu.Unlock()

  g.doCall(c, key, fn)//第一個(gè)進(jìn)來(lái)的協(xié)程來(lái)執(zhí)行這個(gè)函數(shù)

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()//執(zhí)行被包裝的函數(shù)

  c.wg.Done()//執(zhí)行完畢后，就可以通知其他協(xié)程可以拿結(jié)果了

  g.mu.Lock()

if !c.forgotten {//其實(shí)這里是為了保證執(zhí)行完畢之后，對(duì)應(yīng)的key被刪除，Group有一個(gè)方法Forget（key string），可以用來(lái)主動(dòng)刪除key，這里是判斷那個(gè)方法是否被調(diào)用過(guò)，被調(diào)用過(guò)則字段forgotten會(huì)置為true，如果沒有被調(diào)用過(guò)，則在這里把key刪除。

  delete(g.m, key)

  }

  for _, ch := range c.chans {//將執(zhí)行結(jié)果發(fā)送到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 {//加入了超時(shí)機(jī)制

    case <-timeout:

      log.Printf("time out!")

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

    case ret =<- retChan://從chan中取出結(jié)果

      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)//可以看到，每個(gè)等待的協(xié)程，都有一個(gè)結(jié)果channel。從之前的g.doCall里也可以看到，每個(gè)channel都給塞了結(jié)果。為什么不所有協(xié)程共用一個(gè)channel？因?yàn)槟菢泳偷迷赾hannel里塞至少與協(xié)程數(shù)量一樣的結(jié)果數(shù)量，但是你卻無(wú)法保證用戶一個(gè)協(xié)程只讀取一次。

   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
}