syn/singleflight源码阅读

2021-06-21

singleflight是go扩展库提供的一种并发原语，当有大量并发请求时，只允许一个请求去实际调用这个回调函数，等到这个请求返回结果的时候，再把结果返回给其他几个同时调用了相同函数的请求，这样可以减少并发调用的数量。在实际应用中，它能够在一个服务中减少对下游的并发重复请求。一个常见的使用场景是防止缓存击穿

所谓缓存击穿，指的是大量的并发请求同时查询一个缓存Key时，如果这个Key正好过期失效，就会导致大量的请求都打到数据库上，这种现象就叫作缓存击穿。而singleflight能够在有大量针对同一key的请求时，只让一个请求执行去获取数据，而其他协程阻塞等待结果的返回，因此能有效避免这种现象

							 //核心数据结构如下：
type Group struct {
	mu sync.Mutex       //互斥锁
	m  map[string]*call //对于每一个要获取的key有一个对应的call
}
 
type call struct {
	wg sync.WaitGroup   //阻塞这个call的其他调用请求
 
	//调用函数后返回的值和error，只会写一次，并且在wg的Done方法执行后才能被读取
	val interface{}
	err error
 
	forgotten bool		//标识fn方法执行完成之后结果是否立即删除还是保留在singleflight中
 
	dups  int			//同时执行一个key的对应call的协程数量
	chans []chan<- Result
}
 
//通常用在DoChan方法中
type Result struct {
	Val    interface{}
	Err    error
	Shared bool
}


							 //核心使用方法如下：
//fn为回调函数
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)
	}
    
    //查看是否是首次执行key的call方法，否的话则阻塞当前协程，等待其他执行call方法的协程唤醒返回数据
	if c, ok := g.m[key]; ok {
		c.dups++
		g.mu.Unlock()
		c.wg.Wait()

		if e, ok := c.err.(*panicError); ok {
			panic(e)
		} else if c.err == errGoexit {
			runtime.Goexit()
		}
		return c.val, c.err, true
	}
    
    //首次执行call，新建key对应的value，阻塞其他协程
	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
}

可以注意到，Do封锁了其他并发协程调用fn的可能，最终逻辑走向单独需要调用的fn上，但我们需要给它做一些封装，这也就是doCall方法

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
	normalReturn := false	//正常执行fn
	recovered := false		//未正常执行fn，但执行了recover

	defer func() {
		//既没有执行fn也未recover
		if !normalReturn && !recovered {
			c.err = errGoexit
		}

		c.wg.Done()
		g.mu.Lock()
		defer g.mu.Unlock()
        //如果未删除记得删掉对应的call
		if !c.forgotten {
			delete(g.m, key)
		}

		if e, ok := c.err.(*panicError); ok {
			if len(c.chans) > 0 {
				go panic(e)
				select {} 
			} else {
				panic(e)
			}
		} else if c.err == errGoexit {
			//已经退出
		} else {
			//正常返回
			for _, ch := range c.chans {
				ch <- Result{c.val, c.err, c.dups > 0}
			}
		}
	}()

	func() {
		defer func() {
			if !normalReturn {
				//若fn panic了则进行recover，并new一个panic型的error
				if r := recover(); r != nil {
					c.err = newPanicError(r)
				}
			}
		}()
		//这步才是所有的核心操作
		c.val, c.err = fn()
        //若fn 未panic则执行到这步返回
		normalReturn = true
	}()

	if !normalReturn {
		recovered = true
	}
}

当然还有一些其他的方法，如下：

//与Do类似，不同之处在于Do是同步返回结果，DoChan返回一个channel（每个协程都有各自单独的channel），异步将结果返回
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
}


//将key从map中删除
func (g *Group) Forget(key string) {
	g.mu.Lock()
	if c, ok := g.m[key]; ok {
		c.forgotten = true
	}
	delete(g.m, key)
	g.mu.Unlock()
}

大致逻辑图如下：