Go 用 errgroup 管理并发子任务:错误收敛、取消传播与限流并发拉一批数据,你大概率写过这样的代码:开五个 goroutine 各自请求一个接口,用sync.WaitGroup等它们全部结束。跑起来没问题,直到有一天某个接口挂了——其他四个还在傻等,错误也没人收,日志里只留下一条超时。更糟的是,第一个失败之后剩下的请求已经没意义了,却还在白白消耗连接和 CPU。golang.org/x/sync/errgroup就是为解决这类问题而生的:它在 WaitGroup 的基础上,帮你收敛第一个错误、并在出错时自动取消其他子任务。这篇把它的三个实战用法讲透:错误收敛、取消传播、并发限流。先看朴素写法的问题用原生 WaitGroup 并发请求,想拿到「任意一个失败就返回错误」并不容易:funcfetchAll(urls[]string)error{varwg sync.WaitGroupvarmu sync.MutexvarfirstErrerrorfor_,u:rangeurls{wg.Add(1)gofunc(ustring){deferwg.Done()iferr:fetch(u);err!nil{mu.Lock()iffirstErrnil{// 只记第一个错误,得手动加锁firstErrerr}mu.Unlock()}}(u)}wg.Wait()returnfirstErr}问题一眼看不完:要手动加锁保护firstErr、要手动Add/Done、而且某个请求失败后其余请求不会停,continue 跑到底。这些样板代码每写一次都容易漏一个wg.Done()导致死锁。用 errgroup 收敛错误同样的逻辑,errgroup 一把梭:importgolang.org/x/sync/errgroupfuncfetchAll(urls[]string)error{varg errgroup.Groupfor_,u:rangeurls{u:u// 捕获循环变量(Go 1.22 前必须,之后可省)g.Go(func()error{returnfetch(u)// 直接返回 error,g 帮你收第一个})}returng.Wait()// 任意一个非 nil,Wait 就返回它}g.Go接收一个func() error,内部帮你管理 WaitGroup;g.Wait()会阻塞到所有子任务结束,并返回第一个非 nil 的错误。锁、计数器全省了。注意语义:Wait只返回第一个错误,后面的错误会被丢弃。如果你需要所有错误,得自己在闭包里收集(比如塞进一个带锁的 slice),errgroup 不替你做这件事。关键升级:出错就取消其他任务上面的版本虽然收了错,但失败后其余请求仍会跑完。真正的价值在errgroup.WithContext:它派生一个 context,任意子任务返回非 nil 错误时,自动 cancel 这个 context,其他子任务只要监听了 ctx 就能及时退出。funcfetchAll(ctx context.Context,urls[]string)([]string,error){g,ctx:errgroup.WithContext(ctx)// 派生可取消的 ctxresults:make([]string,len(urls))fori,u:rangeurls{i,u:i,u g.Go(func()error{// 把 ctx 透传给下游,失败时这里会随 ctx 一起被取消body,err:fetchWithCtx(ctx,u)iferr!nil{returnerr// 触发 ctx cancel,其他 goroutine 收到 Done}results[i]body// 各写各的下标,无需加锁returnnil})}iferr:g.Wait();err!nil{returnnil,err}returnresults,nil}funcfetchWithCtx(ctx context.Context,urlstring)(string,error){req,_:http.NewRequestWithContext(ctx,http.MethodGet,url,nil)resp,err:http.DefaultClient.Do(req)// ctx 被 cancel 时这里立刻返回 erroriferr!nil{return,err}deferresp.Body.Close()b,err:io.ReadAll(resp.Body)returnstring(b),err}这里有个容易忽略的细节:results[i] body每个 goroutine 写不同下标,读写不重叠,所以不需要加锁。这是并发写 slice 安全的少数场景之一——前提是长度预分配好、各 goroutine 下标互不相同。要让取消真正生效,子任务里的阻塞操作必须监听 ctx。HTTP 用NewRequestWithContext,数据库用QueryContext,自己的循环用select { case -ctx.Done(): return ctx.Err() ... }。如果下游不吃 ctx,cancel 就是一纸空文,任务照样跑到底。用 SetLimit 限流,别一次开几千个 goroutine如果 urls 有几千个,上面的写法会瞬间开几千个 goroutine 和连接,把下游打挂。Go 1.20 起 errgroup 内置了SetLimit,限制同时运行的子任务数:funcfetchAll(ctx context.Context,urls[]string)error{g,ctx:errgroup.WithContext(ctx)g.SetLimit(10)// 最多 10 个 goroutine 并发for_,u:rangeurls{u:u g.Go(func()error{// 超过 10 个时,g.Go 会阻塞直到有空位returnfetchWithCtx2(ctx,u)})}returng.Wait()}SetLimit(n)之后,g.Go在并发数达到上限时会阻塞,直到有子任务结束腾出名额。这等价于自己维护一个容量为 n 的信号量 channel,但代码干净得多。注意SetLimit必须在任何g.Go之前调用,中途改会 panic。如果你不想让g.Go阻塞,而是希望「没名额就跳过」,可以用TryGo,它返回 bool 表示是否成功启动。一个真实的坑:漏传 ctx 导致取消失效最常见的翻车是这样:用了WithContext拿到新 ctx,却在子任务里图省事用了外层的旧 ctx 或context.Background()。g,ctx:errgroup.WithContext(parentCtx)g.Go(func()error{// 错误:用了 parentCtx,errgroup 的 cancel 传不进来returnfetchWithCtx(parentCtx,u)})errgroup.WithContext返回的新ctx 才是挂了 cancel 钩子的那个。子任务必须用这个新 ctx,取消才会传播。记住口诀:g, ctx : errgroup.WithContext(...)之后,组内一律用左边这个新ctx。小结并发子任务要「收敛第一个错误」,别再手写 WaitGroup 锁,用errgroup.Group。需要「一个失败就停掉其他」时用errgroup.WithContext,并把返回的新 ctx透传给每个子任务的阻塞调用。任务数很大时用SetLimit(n)限流,避免 goroutine 与连接爆炸;要非阻塞就用TryGo。各 goroutine 写不同 slice 下标可以免锁,但共享变量仍要自己保护。一句话记忆点:errgroup WaitGroup 第一个错误 自动取消 限流,但取消能不能生效,全看你有没有把新 ctx 传下去。
Go 用 errgroup 管理并发子任务:错误收敛、取消传播与限流并发拉一批数据,你大概率写过这样的代码:开五个 goroutine 各自请求一个接口,用sync.WaitGroup等它们全部结束。跑起来没问题,直到有一天某个接口挂了——其他四个还在傻等,错误也没人收,日志里只留下一条超时。更糟的是,第一个失败之后剩下的请求已经没意义了,却还在白白消耗连接和 CPU。golang.org/x/sync/errgroup就是为解决这类问题而生的:它在 WaitGroup 的基础上,帮你收敛第一个错误、并在出错时自动取消其他子任务。这篇把它的三个实战用法讲透:错误收敛、取消传播、并发限流。先看朴素写法的问题用原生 WaitGroup 并发请求,想拿到「任意一个失败就返回错误」并不容易:funcfetchAll(urls[]string)error{varwg sync.WaitGroupvarmu sync.MutexvarfirstErrerrorfor_,u:rangeurls{wg.Add(1)gofunc(ustring){deferwg.Done()iferr:fetch(u);err!nil{mu.Lock()iffirstErrnil{// 只记第一个错误,得手动加锁firstErrerr}mu.Unlock()}}(u)}wg.Wait()returnfirstErr}问题一眼看不完:要手动加锁保护firstErr、要手动Add/Done、而且某个请求失败后其余请求不会停,continue 跑到底。这些样板代码每写一次都容易漏一个wg.Done()导致死锁。用 errgroup 收敛错误同样的逻辑,errgroup 一把梭:importgolang.org/x/sync/errgroupfuncfetchAll(urls[]string)error{varg errgroup.Groupfor_,u:rangeurls{u:u// 捕获循环变量(Go 1.22 前必须,之后可省)g.Go(func()error{returnfetch(u)// 直接返回 error,g 帮你收第一个})}returng.Wait()// 任意一个非 nil,Wait 就返回它}g.Go接收一个func() error,内部帮你管理 WaitGroup;g.Wait()会阻塞到所有子任务结束,并返回第一个非 nil 的错误。锁、计数器全省了。注意语义:Wait只返回第一个错误,后面的错误会被丢弃。如果你需要所有错误,得自己在闭包里收集(比如塞进一个带锁的 slice),errgroup 不替你做这件事。关键升级:出错就取消其他任务上面的版本虽然收了错,但失败后其余请求仍会跑完。真正的价值在errgroup.WithContext:它派生一个 context,任意子任务返回非 nil 错误时,自动 cancel 这个 context,其他子任务只要监听了 ctx 就能及时退出。funcfetchAll(ctx context.Context,urls[]string)([]string,error){g,ctx:errgroup.WithContext(ctx)// 派生可取消的 ctxresults:make([]string,len(urls))fori,u:rangeurls{i,u:i,u g.Go(func()error{// 把 ctx 透传给下游,失败时这里会随 ctx 一起被取消body,err:fetchWithCtx(ctx,u)iferr!nil{returnerr// 触发 ctx cancel,其他 goroutine 收到 Done}results[i]body// 各写各的下标,无需加锁returnnil})}iferr:g.Wait();err!nil{returnnil,err}returnresults,nil}funcfetchWithCtx(ctx context.Context,urlstring)(string,error){req,_:http.NewRequestWithContext(ctx,http.MethodGet,url,nil)resp,err:http.DefaultClient.Do(req)// ctx 被 cancel 时这里立刻返回 erroriferr!nil{return,err}deferresp.Body.Close()b,err:io.ReadAll(resp.Body)returnstring(b),err}这里有个容易忽略的细节:results[i] body每个 goroutine 写不同下标,读写不重叠,所以不需要加锁。这是并发写 slice 安全的少数场景之一——前提是长度预分配好、各 goroutine 下标互不相同。要让取消真正生效,子任务里的阻塞操作必须监听 ctx。HTTP 用NewRequestWithContext,数据库用QueryContext,自己的循环用select { case -ctx.Done(): return ctx.Err() ... }。如果下游不吃 ctx,cancel 就是一纸空文,任务照样跑到底。用 SetLimit 限流,别一次开几千个 goroutine如果 urls 有几千个,上面的写法会瞬间开几千个 goroutine 和连接,把下游打挂。Go 1.20 起 errgroup 内置了SetLimit,限制同时运行的子任务数:funcfetchAll(ctx context.Context,urls[]string)error{g,ctx:errgroup.WithContext(ctx)g.SetLimit(10)// 最多 10 个 goroutine 并发for_,u:rangeurls{u:u g.Go(func()error{// 超过 10 个时,g.Go 会阻塞直到有空位returnfetchWithCtx2(ctx,u)})}returng.Wait()}SetLimit(n)之后,g.Go在并发数达到上限时会阻塞,直到有子任务结束腾出名额。这等价于自己维护一个容量为 n 的信号量 channel,但代码干净得多。注意SetLimit必须在任何g.Go之前调用,中途改会 panic。如果你不想让g.Go阻塞,而是希望「没名额就跳过」,可以用TryGo,它返回 bool 表示是否成功启动。一个真实的坑:漏传 ctx 导致取消失效最常见的翻车是这样:用了WithContext拿到新 ctx,却在子任务里图省事用了外层的旧 ctx 或context.Background()。g,ctx:errgroup.WithContext(parentCtx)g.Go(func()error{// 错误:用了 parentCtx,errgroup 的 cancel 传不进来returnfetchWithCtx(parentCtx,u)})errgroup.WithContext返回的新ctx 才是挂了 cancel 钩子的那个。子任务必须用这个新 ctx,取消才会传播。记住口诀:g, ctx : errgroup.WithContext(...)之后,组内一律用左边这个新ctx。小结并发子任务要「收敛第一个错误」,别再手写 WaitGroup 锁,用errgroup.Group。需要「一个失败就停掉其他」时用errgroup.WithContext,并把返回的新 ctx透传给每个子任务的阻塞调用。任务数很大时用SetLimit(n)限流,避免 goroutine 与连接爆炸;要非阻塞就用TryGo。各 goroutine 写不同 slice 下标可以免锁,但共享变量仍要自己保护。一句话记忆点:errgroup WaitGroup 第一个错误 自动取消 限流,但取消能不能生效,全看你有没有把新 ctx 传下去。
