概述

singleflight.Group 是 Go 語言擴展包的另一種同步原語，它能夠再一個服務中抑制對下游的多次重復請求。一個比較常見的使用場景是，我們使用 Redis 對數(shù)據(jù)庫中的數(shù)據(jù)進行緩存，發(fā)生緩存擊穿時，大量請求會打到數(shù)據(jù)庫上進而影響服務的尾延時。

而 singleflight.Group 能夠有效地解決這個問題，它能夠限制對同一個鍵值對的多次重復請求，減少對下游的瞬時流量。

在資源的獲取非常昂貴時（例如訪問緩存、數(shù)據(jù)庫），就很適合使用 singleflight.Group 優(yōu)化服務。它的使用方法如下：

type service struct {
    requestGroup singleflight.Group
}
func (s *service) handleRequest(ctx context.Context, request Request) (Response, error) {
    v, err, _ := requestGroup.Do(request.Hash(), func() (interface{}, error) {
        rows, err := // select * from tables
        if err != nil {
            return nil, err
        }
        return rows, nil
    })
    if err != nil {
        return nil, err
    }
    return Response{
        rows: rows,
    }, nil
}

因為請求的哈希在業(yè)務上一般表示相同的請求，所以上述代碼使用它作為請求的鍵。當然，我們也可以選擇其他的字段作為 singleflight.Group.Do 方法的第一個參數(shù)減少重復的請求。

結構體

singleflight.Group 結構體由一個互斥鎖 sync.Mutex 和一個映射表組成，每一個 singleflight.call 結構體都保存了當前調(diào)用對應的信息：

type Group struct {
	mu sync.Mutex
	m  map[string]*call
}
type call struct {
	wg sync.WaitGroup
	val interface{}
	err error
	dups  int
	chans []chan<- Result
}

singleflight.call 結構體中的 val 和 err 字段都只會在執(zhí)行傳入的函數(shù)時賦值一次并在 sync.WaitGroup.Wait 返回時被讀取。

dups 和 chans 兩個字段分別存儲了抑制的請求數(shù)量以及用于同步結果的 Channel。

接口

singleflight.Group求的方法：

• singleflight.Group.Do  — 同步等待的方法；
• singleflight.Group.DoChan — 返回 Channel 異步等待的方法；

這兩個方法在功能上沒有太多的區(qū)別，只是在接口的表現(xiàn)上稍有不同。

每次調(diào)用 singleflight.Group.Do 方法時都會獲取互斥鎖，隨后判斷是否已經(jīng)存在鍵對應的 singleflight.call：

• 當不存在對應的 singleflight.call 時：

  • 初始化一個新的 singleflight.call 指針
  • 增加 sync.WaitGroup 持有的計數(shù)器
  • 將 singleflight.call 指針添加到映射表
  • 釋放持有的互斥鎖
  • 阻塞地調(diào)用 singleflight.Group.doCall 方法等待結果的返回

• 當存在對應的 singleflight.call 時：

  • 增加 dups 計數(shù)器，它表示當前重復的調(diào)用次數(shù)
  • 釋放持有的互斥鎖
  • 通過 sync.WaitGroup.Wait 等待請求的返回

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
    g.mu.Lock()
    if g.m == nil {
            g.m = make(map[string]*call)
    }
    if c, ok := g.m[key]; ok {
            c.dups++
            g.mu.Unlock()
            c.wg.Wait()
            return c.val, c.err, true
    }
    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
}

因為 val 和 err 兩個字段都只會在 singleflight.Group.doCall 方法中賦值，所以當 singleflight.Group.doCall 和 sync.WaitGroup.Wait 返回時，函數(shù)調(diào)用的結果和錯誤都會返回給 singleflight.Group.Do 的調(diào)用方。

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
	c.val, c.err = fn()
	c.wg.Done()
	g.mu.Lock()
	delete(g.m, key)
	for _, ch := range c.chans {
		ch <- Result{c.val, c.err, c.dups > 0}
	}
	g.mu.Unlock()
}

• 行傳入的函數(shù) fn，該函數(shù)的返回值會賦值給 c.val 和 c.err
• 調(diào)用 sync.WaitGroup.Done 方法通知所有等待結果的 Goroutine — 當前函數(shù)已經(jīng)執(zhí)行完成，可以從 call 結構體中取出返回值并返回了
• 獲取持有的互斥鎖并通過管道將信息同步給使用 singleflight.Group.DoChan 方法的 Goroutine

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.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
}

singleflight.Group.Do 和 singleflight.Group.DoChan 分別提供了同步和異步的調(diào)用方式，這讓我們使用起來也更加靈活。

小結

當我們需要減少對下游的相同請求時，可以使用 singleflight.Group 來增加吞吐量和服務質量，不過在使用的過程中我們也需要注意以下的幾個問題：

• singleflight.Group.Do 和 singleflight.Group.DoChan 一個用于同步阻塞調(diào)用傳入的函數(shù)，一個用于異步調(diào)用傳入的參數(shù)并通過 Channel 接收函數(shù)的返回值
• singleflight.Group.Forget 可以通知 singleflight.Group 在持有的映射表中刪除某個鍵，接下來對該鍵的調(diào)用就不會等待前面的函數(shù)返回了
• 一旦調(diào)用的函數(shù)返回了錯誤，所有在等待的 Goroutine 也都會接收到同樣的錯誤

到此這篇關于Go擴展原語之SingleFlight的用法詳解的文章就介紹到這了,更多相關Go SingleFlight內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論