理解 Go singleflight 中的异常处理

sorcererxw /

golang.org/x/sync/singleflight 可以避免多个 Goroutine 同时重复地做同一件事情。

它的源码非常简单,整体流程如下:

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

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

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error) {
	g.mu.Lock()

	if c, ok := g.m[key]; ok {
		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
}

每一个请求进来的时候:

  • 如果是第一次将其封装为一个 call 的对象并放入请求池内
  • 之后所有相同的请求,就是等待原来的 call 结束,并直接使用它的结果

相比起大的流程,singleflight 的 doCall 函数更有意思:

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
	normalReturn := false // 是否成功运行完成
	recovered := false // 是否发生 panic

	defer func() {
		if !normalReturn && !recovered {
			c.err = errGoexit // 既没有成功运行,也没有 panic,那么就只有可能是 runtime.Goexit
		}
		g.mu.Lock()
		defer g.mu.Unlock()

		if e, ok := c.err.(*panicError); ok {
			if len(c.chans) > 0 {
				// 为什么要单开一个进程 panic
				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,使代码更加简洁
		defer func() {
			if !normalReturn {
				if r := recover(); r != nil {
					c.err = newPanicError(r)
				}
			}
		}()

		c.val, c.err = fn()
		normalReturn = true
	}()

	if !normalReturn {
		recovered = true
	}
}
  • 为什么会出现 recovered == false 但是 normalReturn ==false 的情况?

    Go 当中有两种方式可以立即中断 goroutine:

    • panic:可以使用 recover 进行捕获错误并恢复,如果没有 recover 会导致整个进程退出
    • runtime.Goexit:立即结束当前 goroutine 的运行

    两者共同的特点都是:结束之后,它们栈上的 defer 依然会被依次执行。那么就是可以通过在 defer 当中判断函数是否正常运行结束,还是发生了 recover,如果两者均未发生,就是发生 runtime.Goexit。