解决Golang并发工具Singleflight的问题

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++ { // 模拟多协程同时请求
      go func(idx int) { // 注意for的一个小坑
         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对应的value值
func getValueBySingleflight(idx int, cacheKey string) (string, error) {
   log.Printf("idx %v into-cache...", idx)
   // 调用singleflight的Do()方法
   value, _, _ := sf.Do(cacheKey, func() (ret interface{}, err error) {
      log.Printf("idx %v is-setting-cache", idx)
      // 休眠0.1s以捕获并发的相同请求
      time.Sleep(100 * time.Millisecond)
      log.Printf("idx %v set-cache-success!", idx)
      return "myValue", nil
   })
   return value.(string), nil
}

type (
   Group struct { // singleflight实体
      mu sync.Mutex       // 互斥锁
      m  map[string]*call // 懒加载
   }
   call struct {
      wg sync.WaitGroup
      // 存储 调用singleflight.Do()方法返回的结果
      val interface{}
      err error
      // 调用singleflight.Forget(key)时将对应的key从Group.m中删除
      forgotten bool
      // 通俗的理解成singleflight合并的并发请求数
      dups  int
      // 存储 调用singleflight.DoChan()方法返回的结果
      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 { // 检查相同的请求已经是否进入过singleflight
      c.dups++
      g.mu.Unlock()
      c.wg.Wait() // 调用waitGroup的wait()方法阻塞住本次调用，等待第一个进入singleflight的请求执行完毕拿到结果，将本次请求唤醒.
      if e, ok := c.err.(*panicError); ok { //如果调用完成，发生error ，将error上抛
         panic(e)
      } else if c.err == errGoexit {
         runtime.Goexit()
      }
      // 返回调用结果
      return c.val, c.err, true
   }
   c := new(call) // 相同的请求第一次进入singleflight
   c.wg.Add(1)
   g.m[key] = c // new一个call实体，放入singleflight.call这个map
   g.mu.Unlock()
   g.doCall(c, key, fn) //实际执行的函数
   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标志位记为true
   }
   delete(g.m, key)  // Group.m中删除对应的key
   g.mu.Unlock()
}

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
   normalReturn := false
   recovered := false
    //使用双重defer来区分error的类型: panic && runtime.error
   defer func() { 
      if !normalReturn && !recovered {
        // fn()发生了panic且fn()中的panic没有被recover掉
        // errGoexit连接runtime.Goexit错误
         c.err = errGoexit 
      }
      c.wg.Done()
      g.mu.Lock()
      defer g.mu.Unlock()
      if !c.forgotten { // 检查key是否调用了Forget()
         delete(g.m, key)
      }
      if e, ok := c.err.(*panicError); ok {
         // 如果返回的是 panic 错误，为了避免channel被永久阻塞，我们需要确保这个panic无法被recover
         if len(c.chans) > 0 {
            go panic(e)  // panic无法被恢复
            select {} // 阻塞本goroutinue.
         } else {
            panic(e)
         }
      } else {
         // 将结果正常地返回
         for _, ch := range c.chans {
            ch <- Result{c.val, c.err, c.dups > 0}
         }
      }
   }()
   func() {
      defer func() {
         if !normalReturn {
            // 表示fn()发生了panic()
            // 此时与panic相关的堆栈已经被丢弃(调用的fn()) ，无法通过堆栈跟踪去确定error类型
            if r := recover(); r != nil {
               c.err = newPanicError(r) //new一个新的自定义panic err,往第一个defer抛
            }
         }
      }()
     // 执行我们实际的业务逻辑，并将业务方法的返回值赋给singleflight.call
      c.val, c.err = fn()的val和err属性
      // 如果fn()发生panic，normalReturn无法被赋值为true，而是进入doCall()的第二个defer()
      normalReturn = true
   }()
   // 如果normalResult为false时，表示fn()发生了panic
   // 但是执行到了这一步，表示fn()中的panic被recover了
   if !normalReturn {
      recovered = true // recovered标志位置为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无数据，从redis查
    resData, err = srv.rdsRepo.Get()
    if err != nil && err != redis.Nil {
       // redis错误
       log.Fatalln()
       return resData, err
    } else if redis.Nil == err {
           // redis无数据 ，查db
           resData, err, _ = singleFlight.Do(key, func() (interface{}, error) {
           // 构建db查询条件
          searchConn := entity.SearchInfo{}
           //  建议休眠0.1s 捕获0.1s内的重复请求
          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数据,回写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 //协程数
   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的入参key，可以直接使用缓存的key，这样同一个缓存，只有一个协程会去读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 超时控制
      select {
      case <-ctx.Done():
         return "", errors.New("ctx-timeout") // 根据业务逻辑选择上抛 error
      case data, _ := <-ch:
         return data.Val.(string), nil
      default:
      }
   }
}

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