hello，大家好啊，我是小樓。

最近在工作中對 Go 的 singleflight 包做了下增強(qiáng)，解決了一個性能問題，這里記錄下，希望對你也有所幫助。

singleflight 是什么

singleflight 直接翻譯為"單（次）飛（行）"，它是對同一種請求的抑制，保證同一時刻相同的請求只有一個在執(zhí)行，且在它執(zhí)行期間的相同請求都會 Hold 直到執(zhí)行完成，這些 Hold 的請求也使用這次執(zhí)行的結(jié)果。

舉個例子，當(dāng)程序中有讀（如 Redis、MySQL、Http、RPC等）請求，且并發(fā)非常高的情況，使用 singleflight 能得到比較好的效果，它限制了同一時刻只有一個請求在執(zhí)行，也就是并發(fā)永遠(yuǎn)為1。

singleflight 的原理

最初 singleflight 出現(xiàn)在 groupcache 項(xiàng)目中，這個項(xiàng)目也是 Go 團(tuán)隊(duì)所寫，后來該包被移到 Go 源碼中，在 Go 源碼中的版本經(jīng)過幾輪迭代，稍微有點(diǎn)復(fù)雜，我們以最原始的源碼來講解原理，更方便地看清本質(zhì)。

https://github.com/golang/groupcache/blob/master/singleflight/singleflight.go

singleflight 把每次請求定義為 call，每個 call 對象包含了一個 waitGroup，一個 val，即請求的返回值，一個 err，即請求返回的錯誤。

type call struct {
 wg  sync.WaitGroup
 val interface{}
 err error
}

再定義全局的 Group，包含一個互斥鎖 Mutex，一個 key 為 string，value 為 call 的 map。

type Group struct {
 mu sync.Mutex       
 m  map[string]*call
}

Group 對象有一個 Do 方法，其第一個參數(shù)是 string 類型的 key，這個 key 也就是上面說的 map 的 key，相同的 key 標(biāo)志著他們是相同的請求，只有相同的請求會被抑制；第二個參數(shù)是一個函數(shù) fn，這個函數(shù)是真正要執(zhí)行的函數(shù)，例如調(diào)用 MySQL；返回值比較好理解，即最終調(diào)用的返回值和錯誤信息。

func (g *Group) Do(key string, fn func() (interface{}, error)) (interface{}, error) {
 // ①
  g.mu.Lock()
 if g.m == nil {
  g.m = make(map[string]*call)
 }
  // ②
 if c, ok := g.m[key]; ok {
  g.mu.Unlock()
  c.wg.Wait()
  return c.val, c.err
 }
  // ③
 c := new(call)
 c.wg.Add(1)
 g.m[key] = c
 g.mu.Unlock()

 c.val, c.err = fn()
 c.wg.Done()

 g.mu.Lock()
 delete(g.m, key)
 g.mu.Unlock()

 return c.val, c.err
}

將整個代碼分成三塊：

• ① 懶加載方式初始化 map；
• ② 如果當(dāng)前 key 存在，即相同請求正在調(diào)用中，就等它完成，完成后直接使用它的 value 和 error；
• ③ 如果當(dāng)前 key 不存在，即沒有相同請求正在調(diào)用中，就創(chuàng)建一個 call 對象，并把它放進(jìn) map，接著執(zhí)行 fn 函數(shù)，當(dāng)函數(shù)執(zhí)行完喚醒 waitGroup，并刪除 map 相應(yīng)的 key，返回 value 和 error。

讀可以抑制，寫呢？

我們通過上面的介紹能了解，singleflight 能解決并發(fā)讀的問題，但我又遇到一個并發(fā)寫的問題。為了能讓大家快速進(jìn)入狀態(tài)，先花一點(diǎn)篇幅描述一下遇到的實(shí)際問題：

微服務(wù)中的注冊中心想必大家都有所了解，如果不了解，可以去查查相關(guān)概念，或者翻看我以前的文章，老讀者應(yīng)該能發(fā)現(xiàn)我寫了很多相關(guān)的文章。

服務(wù)提供方在注冊之后，會將變更事件推送到消費(fèi)方，推送事件的處理流程是：接收到事件，查詢組裝出最新的數(shù)據(jù)，然后推送給訂閱者。存在兩種情況可能會導(dǎo)致短時間內(nèi)注冊請求非常多，推送事件多會影響整個注冊中心的性能：

• 接口級注冊（類似 Dubbo），每臺機(jī)器會注冊N多次
• 服務(wù)并發(fā)發(fā)布，例如每次發(fā)布重啟100臺機(jī)器，那么注冊的并發(fā)就可能是100

拿到這種問題，第一想到的解法是：合并推送。但，怎么合并呢？

是不是每次推送的時候等一等，等事件都來了再一把推過去就可以了？但等多久呢？什么時候該等呢？粗暴點(diǎn)，每秒鐘推送一次，這樣就能將一秒內(nèi)的時間都聚合，但這會影響推送的時效性，顯然不符合我們精益求精的要求。

直接使用 singleflight，能行嗎？

套用上面 singleflight ，在第一個事件推送過程中，其他相同的事件被 Hold 住，等第一個事件推送完成后，這些 Hold 的事件不再執(zhí)行推送直接返回。

稍微想一下就知道這樣是有問題的，假設(shè)有三個事件 A、B、C，分別對應(yīng)到三個版本的數(shù)據(jù)A1、B1、C1，A 最先到達(dá)，在 A 開始推送后但沒完成時 B、C 事件到達(dá)，A 事件觸發(fā)推送了 A1 版本的數(shù)據(jù)，B、C 事件在 A 事件推送完成后，直接丟棄，最終推送到消費(fèi)者上的數(shù)據(jù)版本為 A1，但我們肯定期望推送的數(shù)據(jù)版本為 C1，畫個圖線感受下：

增強(qiáng)一點(diǎn)點(diǎn) ????

假設(shè)有事件 A、B、C、D 先后到達(dá)，A 事件仍然先正常執(zhí)行推送，在 A 事件推送的時候，B、C、D 事件 Hold 住，當(dāng) A 事件推送完成后，B 事件開始推送，B 事件將把 A 事件推送時期積攢的事件都一起推送掉，即 B、C、D 一次性推送完成。

增強(qiáng)代碼參考

增強(qiáng)的定義為 WriteGroup，借用 singleflight 原先的實(shí)現(xiàn)，具體代碼就不必解讀了，對照上面的例子應(yīng)該很好理解。

package singleflight

import (
 "sync"
)

type WriteGroup struct {
 mu    sync.Mutex
 wgs   map[string]*sync.WaitGroup
 group Group
}

func (g *WriteGroup) Do(key string, fn func() error) error {
 g.mu.Lock()
 if g.wgs == nil {
  g.wgs = make(map[string]*sync.WaitGroup)
 }
 wg, ok := g.wgs[key]
 if !ok {
  wg = &sync.WaitGroup{}
  wg.Add(1)
  g.wgs[key] = wg
 }
 g.mu.Unlock()

 if !ok {
  err := fn()

  g.mu.Lock()
  wg.Done()
  delete(g.wgs, key)
  g.mu.Unlock()
  return err
 }

 wg.Wait()
 _, err := g.group.Do(key, func() (interface{}, error) {
  return nil, fn()
 })
 return err
}

效果如何？

理論上，如果沒有并發(fā)，事件和以前一樣推送，沒有合并，當(dāng)然這也沒毛病。當(dāng)并發(fā)大于 2 時，開始發(fā)揮威力。在實(shí)際的壓測上，注冊并發(fā) 1500 時，合并的事件達(dá)到 99.9%，效果相當(dāng)炸裂！