Package singleflight provides a duplicate function call suppression

var (
	sfKey1 = "key1"
	wg     *sync.WaitGroup
	sf     singleflight.Group
	nums   = 10
)
func getValueService(key string) { //service
   var val string
   wg = &sync.WaitGroup{}
   wg.Add(nums)
   for idx := 0; idx < nums; idx++ { // 模擬多協(xié)程同時(shí)請求
      go func(idx int) { // 注意for的一個(gè)小坑
         defer wg.Done()
         value, _ := getAndSetCacheNoChan(idx, key) //簡化代碼，不處理error
         log.Printf("request %v get value: %v", idx, value)
         val = value
      }(idx)
   }
   wg.Wait()
   log.Println("val: ", val)
   return
}
// getValueBySingleflight 使用singleflight取cacheKey對應(yīng)的value值
func getValueBySingleflight(idx int, cacheKey string) (string, error) {
   log.Printf("idx %v into-cache...", idx)
   // 調(diào)用singleflight的Do()方法
   value, _, _ := sf.Do(cacheKey, func() (ret interface{}, err error) {
      log.Printf("idx %v is-setting-cache", idx)
      // 休眠0.1s以捕獲并發(fā)的相同請求
      time.Sleep(100 * time.Millisecond)
      log.Printf("idx %v set-cache-success!", idx)
      return "myValue", nil
   })
   return value.(string), nil
}

type (
   Group struct { // singleflight實(shí)體
      mu sync.Mutex       // 互斥鎖
      m  map[string]*call // 懶加載
   }
   call struct {
      wg sync.WaitGroup
      // 存儲(chǔ) 調(diào)用singleflight.Do()方法返回的結(jié)果
      val interface{}
      err error
      // 調(diào)用singleflight.Forget(key)時(shí)將對應(yīng)的key從Group.m中刪除
      forgotten bool
      // 通俗的理解成singleflight合并的并發(fā)請求數(shù)
      dups  int
      // 存儲(chǔ) 調(diào)用singleflight.DoChan()方法返回的結(jié)果
      chans []chan<- Result
   }
   
   Result struct {
      Val    interface{}
      Err    error
      Shared bool
   }
)

func Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool)   
func DoChan(key string, fn func() (interface{}, error)) <-chan Result) 
// 將key從Group.m中刪除
func Forget(key string) 

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
   g.mu.Lock() // 加互斥鎖
   if g.m == nil { // 懶加載map
      g.m = make(map[string]*call)
   }
   if c, ok := g.m[key]; ok { // 檢查相同的請求已經(jīng)是否進(jìn)入過singleflight
      c.dups++
      g.mu.Unlock()
      c.wg.Wait() // 調(diào)用waitGroup的wait()方法阻塞住本次調(diào)用，等待第一個(gè)進(jìn)入singleflight的請求執(zhí)行完畢拿到結(jié)果，將本次請求喚醒.
      if e, ok := c.err.(*panicError); ok { //如果調(diào)用完成，發(fā)生error ，將error上拋
         panic(e)
      } else if c.err == errGoexit {
         runtime.Goexit()
      }
      // 返回調(diào)用結(jié)果
      return c.val, c.err, true
   }
   c := new(call) // 相同的請求第一次進(jìn)入singleflight
   c.wg.Add(1)
   g.m[key] = c // new一個(gè)call實(shí)體，放入singleflight.call這個(gè)map
   g.mu.Unlock()
   g.doCall(c, key, fn) //實(shí)際執(zhí)行的函數(shù)
   return c.val, c.err, c.dups > 0
}

func (g *Group) Forget(key string) {
   g.mu.Lock()
   if c, ok := g.m[key]; ok {
      c.forgotten = true // key的forgotten標(biāo)志位記為true
   }
   delete(g.m, key)  // Group.m中刪除對應(yīng)的key
   g.mu.Unlock()
}

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
   normalReturn := false
   recovered := false
    //使用雙重defer來區(qū)分error的類型: panic && runtime.error
   defer func() { 
      if !normalReturn && !recovered {
        // fn()發(fā)生了panic且fn()中的panic沒有被recover掉
        // errGoexit連接runtime.Goexit錯(cuò)誤
         c.err = errGoexit 
      }
      c.wg.Done()
      g.mu.Lock()
      defer g.mu.Unlock()
      if !c.forgotten { // 檢查key是否調(diào)用了Forget()
         delete(g.m, key)
      }
      if e, ok := c.err.(*panicError); ok {
         // 如果返回的是?panic?錯(cuò)誤，為了避免channel被永久阻塞，我們需要確保這個(gè)panic無法被recover
         if len(c.chans) > 0 {
            go panic(e)  // panic無法被恢復(fù)
            select {} // 阻塞本goroutinue.
         } else {
            panic(e)
         }
      } else {
         // 將結(jié)果正常地返回
         for _, ch := range c.chans {
            ch <- Result{c.val, c.err, c.dups > 0}
         }
      }
   }()
   func() {
      defer func() {
         if !normalReturn {
            // 表示fn()發(fā)生了panic()
            // 此時(shí)與panic相關(guān)的堆棧已經(jīng)被丟棄(調(diào)用的fn()) ，無法通過堆棧跟蹤去確定error類型
            if r := recover(); r != nil {
               c.err = newPanicError(r) //new一個(gè)新的自定義panic err,往第一個(gè)defer拋
            }
         }
      }()
     // 執(zhí)行我們實(shí)際的業(yè)務(wù)邏輯，并將業(yè)務(wù)方法的返回值賦給singleflight.call
      c.val, c.err = fn()的val和err屬性
      // 如果fn()發(fā)生panic，normalReturn無法被賦值為true，而是進(jìn)入doCall()的第二個(gè)defer()
      normalReturn = true
   }()
   // 如果normalResult為false時(shí)，表示fn()發(fā)生了panic
   // 但是執(zhí)行到了這一步，表示fn()中的panic被recover了
   if !normalReturn {
      recovered = true // recovered標(biāo)志位置為true
   }
}

func (s Service) getDataBySingleFlight(ctx  context.Context) (entity.List, error) {
    // 1. 從localCache查
    resData, err := local_cache.Get(ctx, key)
    if err != nil {
       log.Fatalln()
       return resData, err
    }
    if resData != nil {
       return resData, nil
    }
    // 2. localCache無數(shù)據(jù)，從redis查
    resData, err = srv.rdsRepo.Get()
    if err != nil && err != redis.Nil {
       // redis錯(cuò)誤
       log.Fatalln()
       return resData, err
    } else if redis.Nil == err {
           // redis無數(shù)據(jù) ，查db
           resData, err, _ = singleFlight.Do(key, func() (interface{}, error) {
           // 構(gòu)建db查詢條件
          searchConn := entity.SearchInfo{}
           //  建議休眠0.1s 捕獲0.1s內(nèi)的重復(fù)請求
          time.Sleep(100 * time.Millisecond)
           // 4. 查db
          data, err := srv.dBRepo.GetByConn(ctx, searchConn)
          if err != nil {
             log.Fatalln()
             return data, err
          }
           // 5. 回寫localCache && redisCache
          err = local_cache.Set(ctx, data)
          if err != nil {
             log.Fatalln()
          }
          err = srv.rdsRepo.Set(ctx, data)
          if err != nil {
             log.Fatalln()
          }
      // 返回db數(shù)據(jù),回寫cache的error不上拋
      return data, nil
   })
   return resData, err
}
return resData, nil

func enterGetAndSetCacheWithChan(ctx context.Context, key string) (str string, err error) {
   tag := "enterGetAndSetCacheWithChan"
   sonCtx, _ := context.WithTimeout(ctx, 2 * time.Second)
   val := ""
   nums := 10 //協(xié)程數(shù)
   wg = &sync.WaitGroup{}
   wg.Add(nums)
   for idx := 0; idx < nums; idx++ {
      go func() {
         defer wg.Done()
         val, err = getAndSetCacheWithChan(sonCtx, idx, key)
         if err != nil {
            log.Printf("err:[%+v]", err)
            return
         }
         str = val
      }()
   }
   wg.Wait()
   log.Printf("tag:[%s] val:[%s]", tag, val)
   return
}
func getAndSetCacheWithChan(ctx context.Context, idx int, cacheKey string) (string, error) {
   tag := "getAndSetCacheWithChan"
   log.Printf("tag: %s ;idx %d into-cache...", tag, idx)
   ch := sf.DoChan(cacheKey, func() (ret interface{}, err error) { // do的入?yún)ey，可以直接使用緩存的key，這樣同一個(gè)緩存，只有一個(gè)協(xié)程會(huì)去讀DB
      log.Printf("idx %v is-setting-cache", idx)
      time.Sleep(100 * time.Millisecond)
      log.Printf("idx %v set-cache-success!", idx)
      return "myValue", nil
   })
   for { // 選擇 context + select 超時(shí)控制
      select {
      case <-ctx.Done():
         return "", errors.New("ctx-timeout") // 根據(jù)業(yè)務(wù)邏輯選擇上拋 error
      case data, _ := <-ch:
         return data.Val.(string), nil
      default:
      }
   }
}

go func() {
       time.Sleep(100 * time.Millisecond)
       g.Forget(key)
   }()