Golang源碼分析之golang/sync之singleflight

更新時間：2022年11月07日 13:15:58 作者：pbrong

golang/sync庫拓展了官方自帶的sync庫,提供了errgroup、semaphore、singleflight及syncmap四個包,本次先分析第一個包errgroup的源代碼,下面這篇文章主要給大家介紹了關(guān)于Golang源碼分析之golang/sync之singleflight的相關(guān)資料,需要的朋友可以參考下

1.背景

1.1. 項目介紹

golang/sync庫拓展了官方自帶的sync庫，提供了errgroup、semaphore、singleflight及syncmap四個包，本次分析singlefliht的源代碼。
singlefliht用于解決單機協(xié)程并發(fā)調(diào)用下的重復調(diào)用問題，常與緩存一起使用，避免緩存擊穿。

1.2.使用方法

go get -u golang.org/x/sync

核心API：Do、DoChan、Forget
Do：同一時刻對某個Key方法的調(diào)用，只能由一個協(xié)程完成，其余協(xié)程阻塞直到該協(xié)程執(zhí)行成功后，直接獲取其生成的值，以下是一個避免緩存擊穿的常見使用方法：

func main() {
   var flight singleflight.Group
   var errGroup errgroup.Group

   // 模擬并發(fā)獲取數(shù)據(jù)緩存
   for i := 0; i < 10; i++ {
      i := i
      errGroup.Go(func() error {
         fmt.Printf("協(xié)程%v準備獲取緩存\n", i)
         v, err, shared := flight.Do("getCache", func() (interface{}, error) {
            // 模擬獲取緩存操作
            fmt.Printf("協(xié)程%v正在讀數(shù)據(jù)庫獲取緩存\n", i)
            time.Sleep(100 * time.Millisecond)
            fmt.Printf("協(xié)程%v讀取數(shù)據(jù)庫生成緩存成功\n", i)
            return "mockCache", nil
         })
         if err != nil {
            fmt.Printf("err = %v", err)
            return err
         }
         fmt.Printf("協(xié)程%v獲取緩存成功, v = %v, shared = %v\n", i, v, shared)
         return nil
      })
   }
   if err := errGroup.Wait(); err != nil {
      fmt.Printf("errGroup wait err = %v", err)
   }
}
// 輸出：只有0號協(xié)程實際生成了緩存，其余協(xié)程讀取生成的結(jié)果
協(xié)程0準備獲取緩存
協(xié)程4準備獲取緩存
協(xié)程3準備獲取緩存
協(xié)程2準備獲取緩存
協(xié)程6準備獲取緩存
協(xié)程5準備獲取緩存
協(xié)程7準備獲取緩存
協(xié)程1準備獲取緩存
協(xié)程8準備獲取緩存
協(xié)程9準備獲取緩存
協(xié)程0正在讀數(shù)據(jù)庫獲取緩存
協(xié)程0讀取數(shù)據(jù)庫生成緩存成功
協(xié)程0獲取緩存成功, v = mockCache, shared = true
協(xié)程8獲取緩存成功, v = mockCache, shared = true
協(xié)程2獲取緩存成功, v = mockCache, shared = true
協(xié)程6獲取緩存成功, v = mockCache, shared = true
協(xié)程5獲取緩存成功, v = mockCache, shared = true
協(xié)程7獲取緩存成功, v = mockCache, shared = true
協(xié)程9獲取緩存成功, v = mockCache, shared = true
協(xié)程1獲取緩存成功, v = mockCache, shared = true
協(xié)程4獲取緩存成功, v = mockCache, shared = true
協(xié)程3獲取緩存成功, v = mockCache, shared = true

DoChan：將執(zhí)行結(jié)果返回到通道中，可通過監(jiān)聽通道結(jié)果獲取方法執(zhí)行值，這個方法相較于Do來說的區(qū)別是執(zhí)行DoChan后不會阻塞到其中一個協(xié)程完成任務，而是異步執(zhí)行任務，最后需要結(jié)果時直接從通道中獲取，避免長時間等待。

func testDoChan() {
   var flight singleflight.Group
   var errGroup errgroup.Group

   // 模擬并發(fā)獲取數(shù)據(jù)緩存
   for i := 0; i < 10; i++ {
      i := i
      errGroup.Go(func() error {
         fmt.Printf("協(xié)程%v準備獲取緩存\n", i)
         ch := flight.DoChan("getCache", func() (interface{}, error) {
            // 模擬獲取緩存操作
            fmt.Printf("協(xié)程%v正在讀數(shù)據(jù)庫獲取緩存\n", i)
            time.Sleep(100 * time.Millisecond)
            fmt.Printf("協(xié)程%v讀取數(shù)據(jù)庫獲取緩存成功\n", i)
            return "mockCache", nil
         })
         res := <-ch
         if res.Err != nil {
            fmt.Printf("err = %v", res.Err)
            return res.Err
         }
         fmt.Printf("協(xié)程%v獲取緩存成功, v = %v, shared = %v\n", i, res.Val, res.Shared)
         return nil
      })
   }
   if err := errGroup.Wait(); err != nil {
      fmt.Printf("errGroup wait err = %v", err)
   }
}
// 輸出結(jié)果
協(xié)程9準備獲取緩存
協(xié)程0準備獲取緩存
協(xié)程1準備獲取緩存
協(xié)程6準備獲取緩存
協(xié)程5準備獲取緩存
協(xié)程2準備獲取緩存
協(xié)程7準備獲取緩存
協(xié)程8準備獲取緩存
協(xié)程4準備獲取緩存
協(xié)程9正在讀數(shù)據(jù)庫獲取緩存
協(xié)程9讀取數(shù)據(jù)庫獲取緩存成功
協(xié)程3準備獲取緩存
協(xié)程3獲取緩存成功, v = mockCache, shared = true
協(xié)程8獲取緩存成功, v = mockCache, shared = true
協(xié)程0獲取緩存成功, v = mockCache, shared = true
協(xié)程1獲取緩存成功, v = mockCache, shared = true
協(xié)程6獲取緩存成功, v = mockCache, shared = true
協(xié)程5獲取緩存成功, v = mockCache, shared = true
協(xié)程2獲取緩存成功, v = mockCache, shared = true
協(xié)程7獲取緩存成功, v = mockCache, shared = true
協(xié)程4獲取緩存成功, v = mockCache, shared = true
協(xié)程9獲取緩存成功, v = mockCache, shared = true

2.源碼分析

2.1.項目結(jié)構(gòu)

singleflight.go：核心實現(xiàn)，提供相關(guān)API
singleflight_test.go：相關(guān)API單元測試

2.2.數(shù)據(jù)結(jié)構(gòu)

singleflight.go

// singleflight.Group
type Group struct {
   mu sync.Mutex       // map的鎖
   m  map[string]*call // 保存每個key的調(diào)用
}

// 一次Do對應的響應結(jié)果
type Result struct {
   Val    interface{}
   Err    error
   Shared bool
}

// 一個key會對應一個call
type call struct {
   wg sync.WaitGroup
   val interface{} // 保存調(diào)用的結(jié)果
   err error       // 調(diào)用出現(xiàn)的err
   // 該call被調(diào)用的次數(shù)
   dups  int
   // 每次DoChan時都會追加一個chan在該列表
   chans []chan<- Result
}

2.3.API代碼流程

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

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
   g.mu.Lock()
   if g.m == nil {
      // 第一次執(zhí)行Do的時候創(chuàng)建map
      g.m = make(map[string]*call)
   }
   // 已經(jīng)存在該key，對應后續(xù)的并發(fā)調(diào)用
   if c, ok := g.m[key]; ok {
      // 執(zhí)行次數(shù)自增
      c.dups++
      g.mu.Unlock()
      // 等待執(zhí)行fn的協(xié)程完成
      c.wg.Wait()
      // ...
      // 返回執(zhí)行結(jié)果
      return c.val, c.err, true
   }
   
   // 不存在該key，說明第一次調(diào)用，初始化一個call
   c := new(call)
   // wg添加1，后續(xù)其他協(xié)程在該wg上阻塞
   c.wg.Add(1)
   // 保存key和call的關(guān)系
   g.m[key] = c
   g.mu.Unlock()
   // 真正執(zhí)行fn函數(shù)
   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)) {
   normalReturn := false
   recovered := false

   // 第三步、最后的設置和清理工作
   defer func() {
      // ...
      g.mu.Lock()
      defer g.mu.Unlock()
      // 執(zhí)行完成，調(diào)用wg.Done，其他協(xié)程此時不再阻塞，讀到fn執(zhí)行結(jié)果
      c.wg.Done()
      // 二次校驗map中key的值是否為當前call，并刪除該key
      if g.m[key] == c {
         delete(g.m, key)
      }
      // ...
      // 如果c.chans存在，則遍歷并寫入執(zhí)行結(jié)果
      for _, ch := range c.chans {
          ch <- Result{c.val, c.err, c.dups > 0}
        }
      }
   }()

   // 第一步、執(zhí)行fn獲取結(jié)果
   func() {
      // 3、如果fn執(zhí)行過程中panic，將c.err設置為PanicError
      defer func() {
         if !normalReturn {
            if r := recover(); r != nil {
               c.err = newPanicError(r)
            }
         }
      }()
      // 1、執(zhí)行fn，獲取到執(zhí)行結(jié)果
      c.val, c.err = fn()
      // 2、設置正常返回結(jié)果標識
      normalReturn = true
   }()

   // 第二步、fn執(zhí)行出錯，將recovered標識設置為true
   if !normalReturn {
      recovered = true
   }
}

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

func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result {
   // 一次調(diào)用對應一個chan
   ch := make(chan Result, 1)
   g.mu.Lock()
   if g.m == nil {
      // 第一次調(diào)用，初始化map
      g.m = make(map[string]*call)
   }
   // 后續(xù)調(diào)用，已存在key
   if c, ok := g.m[key]; ok {
      // 調(diào)用次數(shù)自增
      c.dups++
      // 將chan添加到chans列表
      c.chans = append(c.chans, ch)
      g.mu.Unlock()
      // 直接返回chan，不等待fn執(zhí)行完成
      return ch
   }

   // 第一次調(diào)用，初始化call及chans列表
   c := &call{chans: []chan<- Result{ch}}
   // wg加一
   c.wg.Add(1)
   // 保存key及call的關(guān)系
   g.m[key] = c
   g.mu.Unlock()

   // 異步執(zhí)行fn函數(shù)
   go g.doCall(c, key, fn)

   // 直接返回該chan
   return ch
}

3.總結(jié)

singleflight經(jīng)常和緩存獲取配合使用，可以緩解緩存擊穿問題，避免同一時刻單機大量的并發(fā)調(diào)用獲取數(shù)據(jù)庫構(gòu)建緩存
singleflight的實現(xiàn)很精簡，核心流程就是使用map保存每次調(diào)用的key與call的映射關(guān)系，每個call中通過wg控制只存在一個協(xié)程執(zhí)行fn函數(shù)，其他協(xié)程等待執(zhí)行完成后，直接獲取執(zhí)行結(jié)果，在執(zhí)行完成后會刪去map中的key
singleflight的Do方法會阻塞直到fn執(zhí)行完成，DoChan方法不會阻塞，而是異步執(zhí)行fn，并通過通道來實現(xiàn)結(jié)果的通知

到此這篇關(guān)于Golang源碼分析之golang/sync之singleflight的文章就介紹到這了,更多相關(guān)Golang源碼分析singleflight內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: