Go底層之singleflight包原理分析

更新時間：2025年06月25日 16:46:16 作者：在成都搬磚的鴨鴨

這篇文章主要介紹了Go底層之singleflight包原理,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

背景

在處理同一時刻接口的并發(fā)請求時，常見的有這幾種情況：

一個請求正在執(zhí)行，相同的其它請求等待順序執(zhí)行，使用互斥鎖就能完成、一個請求正在執(zhí)行，相同的其它請求都丟棄、一個請求正在執(zhí)行，相同的其它請求等待拿取相同的結(jié)果。

使用singleflight包就能達到一個請求正在執(zhí)行，相同的其它請求過來等待第一個請求執(zhí)行完，然后共享第一個請求的結(jié)果，在處理并發(fā)場景時非常好用。

下載

go get -u golang.org/x/sync/singleflight

原理解釋

singleflight底層結(jié)構(gòu)：

type Group struct {
	mu sync.Mutex       //保護map對象m的并發(fā)安全
	m  map[string]*call //key-請求的唯一標識，val-請求唯一標識對應要執(zhí)行的函數(shù)
}

call底層結(jié)構(gòu)：

type call struct {
	wg sync.WaitGroup //用來阻塞相同標識對應的請求中的第一個請求之外的請求
	val interface{} //第一個請求的執(zhí)行結(jié)果
	err error //第一個請求返回的錯誤
	dups  int //第一個請求之外的其它請求數(shù)
	chans []chan<- Result //請求結(jié)果寫入通道
}

關(guān)鍵函數(shù)：

//
// Do
//  @Description: 一個唯一標識對應的請求在執(zhí)行過程中，相同唯一標識對應的請求會被阻塞，等待第一個請求執(zhí)行完并共享結(jié)果
//  @receiver g 
//  @param key 請求唯一標識
//  @param fn 請求要執(zhí)行的函數(shù)
//  @return v 請求要執(zhí)行函數(shù)返回的結(jié)果
//  @return err 請求要執(zhí)行的函數(shù)返回的錯誤
//  @return shared 是否有多個請求共享結(jié)果
//
func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
	g.mu.Lock() //保護map對象m的并發(fā)安全
	if g.m == nil {
		g.m = make(map[string]*call) //初始化m對象
	}
	if c, ok := g.m[key]; ok { //map中key存在說明這個key對應的請求正在執(zhí)行中，這次請求不是第一個請求
		c.dups++ //等待共享結(jié)果數(shù)+1
		g.mu.Unlock()
		c.wg.Wait() //阻塞等待第一個請求執(zhí)行完

		if e, ok := c.err.(*panicError); ok {
			panic(e)
		} else if c.err == errGoexit {
			runtime.Goexit()
		}
		return c.val, c.err, true //返回第一個請求的執(zhí)行結(jié)果
	}
    
    //第一個請求的處理邏輯
	c := new(call)
	c.wg.Add(1) //計數(shù)+1
	g.m[key] = c //唯一標識關(guān)聯(lián)對應的函數(shù)對象
	g.mu.Unlock()

	g.doCall(c, key, fn) //執(zhí)行請求對應的函數(shù)
	return c.val, c.err, c.dups > 0 //返回執(zhí)行結(jié)果
}

上面Do函數(shù)中g(shù).doCall函數(shù)也需要大概理解一下，就是會將key對應函數(shù)的執(zhí)行結(jié)果寫的call對象c里，然后清空wg計數(shù)，相同key對應的其它請求就會跳出c.wg.Wait()阻塞，直接從call對象c中讀取第一個請求的執(zhí)行結(jié)果和錯誤信息并返回，源碼如下：

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
	normalReturn := false
	recovered := false
	
	defer func() { //fn函數(shù)執(zhí)行完之后執(zhí)行
		if !normalReturn && !recovered {
			c.err = errGoexit
		}

		g.mu.Lock()
		defer g.mu.Unlock()
		c.wg.Done() //釋放計數(shù)
		if g.m[key] == c {
			delete(g.m, key) //刪除此次請求的唯一標識相關(guān)信息
		}

		if e, ok := c.err.(*panicError); ok {
			if len(c.chans) > 0 {
				go panic(e)
				select {} 
			} else {
				panic(e)
			}
		} else if c.err == errGoexit {
			// Already in the process of goexit, no need to call again
		} else {
			for _, ch := range c.chans { //執(zhí)行結(jié)果寫入通道
				ch <- Result{c.val, c.err, c.dups > 0}
			}
		}
	}()

	func() {
		defer func() {
			if !normalReturn {
				if r := recover(); r != nil {
					c.err = newPanicError(r)
				}
			}
		}()

		c.val, c.err = fn() //執(zhí)行fn函數(shù)
		normalReturn = true
	}()

	if !normalReturn {
		recovered = true
	}
}

singleflight中還提供了與Do函數(shù)功能相同的函數(shù)DoChan函數(shù)，唯一區(qū)別就是將請求對應的函數(shù)執(zhí)行結(jié)果放到通道中進行返回，這兩函數(shù)一個用于同步場景，一個用于異步場景。還有一個Forget函數(shù)：

func (g *Group) Forget(key string) {
	g.mu.Lock()
	delete(g.m, key) //刪除map中的key，相同key對應請求進來會重新執(zhí)行，不等待第一個key對應請求的執(zhí)行結(jié)果
	g.mu.Unlock()
}

代碼示例

示例如下：

func main() {
	var singleFlight singleflight.Group //初始化一個單次執(zhí)行對象
	var count uint64                    //用于測試是否被修改

	//為了并發(fā)執(zhí)行，這里測試唯一標識都為xxx
	go func() {
		val1, _, shared1 := singleFlight.Do("xxx", func() (interface{}, error) {
			logger.Info("first count +1")

			atomic.AddUint64(&count, 1) //第一次執(zhí)行，將count+1

			time.Sleep(5 * time.Second) //增加第一次執(zhí)行時間

			return count, nil
		})

		//打印第一次執(zhí)行結(jié)果
		logger.Info("first count info", zap.Any("val1", val1), zap.Bool("shared1", shared1), zap.Uint64("count", count))
	}()

	time.Sleep(2 * time.Second) //為了防止下面的Do函數(shù)先執(zhí)行

	val2, _, shared2 := singleFlight.Do("xxx", func() (interface{}, error) {
		logger.Info("second count +1")

		atomic.AddUint64(&count, 1) //第2次執(zhí)行count+1

		return count, nil
	})

	//打印第二次執(zhí)行結(jié)果
	logger.Info("second count info", zap.Any("val2", val2), zap.Bool("shared2", shared2), zap.Uint64("count", count))
}

控制臺輸出：

$ go run ./singlefight_demo/main.go
[2025-01-09 17:08:11.169] | INFO  | Goroutine:6  | [singlefight_demo/main.go:19] | first count +1
[2025-01-09 17:08:16.261] | INFO  | Goroutine:6  | [singlefight_demo/main.go:28] | first count info | {"val1": 1, "shared1": true, "count": 1}
[2025-01-09 17:08:16.261] | INFO  | Goroutine:1  | [singlefight_demo/main.go:41] | second count info | {"val2": 1, "shared2": true, "count": 1}