京公网安备11010502063487号errgroup 是并发任务的常用工具:第一个错误取消其他任务、限制并发数。但它到底怎么做的、有哪些容易踩的坑?这篇用真实运行回答:取消要多久、丢了几个错误、panic 去哪了、ctx 什么时候被取消,再读一遍几十行的实现。
golang.org/x/sync v0.24.0
errgroup 在 golang.org/x/sync 里,不在标准库。本文所有结果来自 v0.24.0,不同版本的行为(尤其是 panic 的处理)可能不同。
任务 1 在 51 ms 发现 ctx 已取消:任务 2 失败 任务 5 在 51 ms 发现 ctx 已取消:任务 2 失败 任务 3 在 51 ms 发现 ctx 已取消:任务 2 失败 任务 4 在 51 ms 发现 ctx 已取消:任务 2 失败 g.Wait() 在 51 ms 返回:任务 2 失败 只用 WaitGroup(没有取消):wg.Wait() 在 1001 ms 才返回,第一个错误是:任务 2 失败
任务 2 出错后,其余任务在 51 ms 就看到了 ctx.Done();同样的任务只用 WaitGroup 要 1001 ms。取消不是强杀,任务必须在 select 里监听 ctx.Done()。
3 个任务先后都出错(20/40/60 ms),g.Wait() 返回:错误 1
3 个任务都出错,Wait 只返回最先出错的那个。如果后面的错误也需要记录,要在任务里自己打日志或收集。
SetLimit(3),12 个各 30 ms 的任务:同时运行的最大个数 = 3,总用时 124 ms,调用 g.Go 累计阻塞 93 ms SetLimit(1) 且已有一个任务在跑时,TryGo 返回 false(不阻塞)
上限 3 时同时运行的最大个数正好是 3,12 个 30 ms 的任务总用时 124 ms;超过上限时 g.Go 会阻塞调用者(累计阻塞 93 ms),要不阻塞就用 TryGo(返回 false)。
进程退出码 2(崩溃);panic: 任务里 panic 了 | <gtmp>/eg1/main.go:89 +0x2c | <gtmp>/_gopath/pkg/mod/golang.org/x/sync@v0.24.0/errgroup/errgroup.go:93 +0x4c | <gtmp>/_gopath/pkg/mod/golang.org/x/sync@v0.24.0/errgroup/errgroup.go:78 +0x98
任务里的 panic 发生在任务自己的 goroutine,在 Wait 所在 goroutine 里 defer recover() 没有用,整个进程退出(退出码 2)。源码注释特意解释了不传播的理由。需要容错就在任务函数里自己 recover。
Wait 返回之后 ctx.Err() = context canceled(WithContext 的 ctx 在 Wait 返回时被取消,即使没有错误也会取消) 所有任务都成功时,Wait 返回之后 ctx.Err() = context canceled
WithContext 返回的 ctx 在第一个错误或 Wait 返回时被取消——实验里即使所有任务成功,Wait 返回后它也是 context canceled。不要在 Wait 之后继续用它。
type Group struct {
cancel func(error)
wg sync.WaitGroup
sem chan token
errOnce sync.Once
err error
}
func (g *Group) Go(f func() error) {
if g.sem != nil {
g.sem <- token{}
}
g.wg.Add(1)
go func() {
defer g.done()
// It is tempting to propagate panics from f()
// up to the goroutine that calls Wait, but
// it creates more problems than it solves:
// - it delays panics arbitrarily,
// making bugs harder to detect;
// - it turns f's panic stack into a mere value,
// hiding it from crash-monitoring tools;
// - it risks deadlocks that hide the panic entirely,
// if f's panic leaves the program in a state
// that prevents the Wait call from being reached.
// See #53757, #74275, #74304, #74306.
if err := f(); err != nil {
g.errOnce.Do(func() {
g.err = err
if g.cancel != nil {
g.cancel(g.err)
}
})
}
}()
}
核心就是 WaitGroup、一个当信号量的 channel、一个 sync.Once 记录第一个错误并 cancel。读源码时那段关于 panic 的注释最值得看。