golang-ErrGroup用法以及源码解读笔记

介绍

ErrGroup可以并发执行多个goroutine，并可以很方便的处理错误

与sync.WaitGroup相比

1. 错误处理
   1. sync.WaitGroup只负责等待goroutine执行完成，而不处理返回值或者错误
   2. errgroup.Group目前虽然不能直接处理函数的返回值或错误。但是当goroutine返回错误的时候，可以取消正在运行的其他goroutine，在Wait方法中返回第一个非nil的错误
2. 上下文取消
   1. errgroup.Group可以与context配合，在一个goroutine出现错误的时候，自动取消其他的goroutine
3. 简化并发编程
   1. errgroup可以减少错误处理的样板代码，开发者不需要手动处理管理错误值和同步逻辑
4. 限制并发数量
   1. errgroup提供便捷的接口来限制并发goroutine的数量，避免过载

api

WithContext

func WithContext(ctx context.Context) (*Group, context.Context)

返回一个新的Group和一个从ctx派生的关联context

传递给Go(func()error)返回到第一个非nil错误，或者Wait第一次返回时，派生的context被取消，先发生者为主

Go

func (g *Group) Go(f func() error)

Go将创建或复用新的goroutine运行给定的任务，对Go()的第一次调用必须先于Wait()。它会阻塞直到新的goroutine可以添加。goroutine的数量不会超过配置的限制

SetLimit

func (g *Group) SetLimit(n int)

将该Group中活动的goroutine的数量限制最多为n，赋值表示没有限制，0限制任何新的goroutine被添加

任何对Go()的后续调用都会被阻塞，直到它可以添加一个获得的goroutine而不超过配置的限制

当组内任何goroutine处于活动状态时，限制不能被修改

TryGo

func (g *Group) TryGo(f func() error) bool

当Group内的goroutine数量小于配置限制时，TryGo才会在goroutine中调用给定的函数

返回值报告goroutine是否启动

Wait

func (g *Group) Wait() error

Wait阻塞，直到所有的函数调用都返回

使用示例

基本使用

package mainimport ("fmt""net/http""golang.org/x/sync/errgroup"
)func main() {g := new(errgroup.Group)var urls = []string{"http://www.golang.org/","http://www.google.com/","http://www.somestupidname.com/",}for _, url := range urls {// Launch a goroutine to fetch the URL.url := url // https://golang.org/doc/faq#closures_and_goroutinesg.Go(func() error {// Fetch the URL.resp, err := http.Get(url)if err != nil {return err}defer resp.Body.Close()fmt.Printf("fetch url %s status %s\n", url, resp.Status)return nil })}// Wait for all HTTP fetches to complete.if err := g.Wait(); err == nil {fmt.Println("Successfully fetched all URLs.")}
}

开启对应数量的goroutine并发访问URL，出现一个错误，主goroutine直接退出，还在访问的不会被取消

上下文取消

package mainimport ("context""fmt""net/http""sync""golang.org/x/sync/errgroup"
)func main() {var urls = []string{"http://www.golang.org/","http://www.google.com/","http://www.somestupidname.com/", // 这是一个错误的 URL，会导致任务失败}// 创建一个带有 context 的 errgroup// 任何一个 goroutine 返回非 nil 的错误，或 Wait() 等待所有 goroutine 完成后，context 都会被取消g, ctx := errgroup.WithContext(context.Background())// 创建一个 map 来保存结果var result sync.Mapfor _, url := range urls {// 使用 errgroup 启动一个 goroutine 来获取 URLg.Go(func() error {req, err := http.NewRequestWithContext(ctx, "GET", url, nil)if err != nil {return err // 发生错误，返回该错误}// 发起请求resp, err := http.DefaultClient.Do(req)if err != nil {return err // 发生错误，返回该错误}defer resp.Body.Close()// 保存每个 URL 的响应状态码result.Store(url, resp.Status)return nil // 返回 nil 表示成功})}// 等待所有 goroutine 完成并返回第一个错误（如果有）if err := g.Wait(); err != nil {fmt.Println("Error: ", err)}// 所有 goroutine 都执行完成，遍历并打印成功的结果result.Range(func(key, value any) bool {fmt.Printf("fetch url %s status %s\n", key, value)return true})
}

创建对应数量的goroutine并发访问URL，如果出现一个错误的话，会取消其他goroutine访问的URL

限制并发数量

package mainimport ("fmt""time""golang.org/x/sync/errgroup"
)func main() {// 创建一个 errgroup.Groupvar g errgroup.Group// 设置最大并发限制为 3g.SetLimit(3)// 启动 10 个 goroutinefor i := 1; i <= 10; i++ {g.Go(func() error {// 打印正在运行的 goroutinefmt.Printf("Goroutine %d is starting\n", i)time.Sleep(2 * time.Second) // 模拟任务耗时fmt.Printf("Goroutine %d is done\n", i)return nil})}// 等待所有 goroutine 完成if err := g.Wait(); err != nil {fmt.Printf("Encountered an error: %v\n", err)}fmt.Println("All goroutines complete.")
}

限制启动的goroutine数量，防止过载

源码

数据结构

type Group struct { // 可为零值cancel func(error) // context的取消函数wg sync.WaitGroup sem chan token // 信号channel，用来控制携程并发数量errOnce sync.Once // 确保错误值处理一次err     error // 记录子协程集中返回的第一个错误
}type token struct{}

SetLimit

限制该group中活动的协程数量最多为n, 负数表示没有限制

任何后续对 Go 方法的调用都将阻塞，直到不超过限额的情况下添加活动协程

在 Group 中存在任何活动的协程时，限制不得修改

n == 0 为导致死锁

func (g *Group) SetLimit(n int) {if n < 0 {g.sem = nilreturn}// 如果存在活动的协程，调用此方法会panicif len(g.sem) != 0 { panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem)))}g.sem = make(chan token, n)
}

Go

Go()会在新的协程中调用给定的函数

它会阻塞，知道可以在不超过配置的活跃协程数量限制的情况下添加新的协程

首次返回非nil的错误的调用会取消该Group的context（如何不为nil）

func (g *Group) Go(f func() error) {// 限制活跃的协程数量if g.sem != nil {g.sem <- token{}}g.wg.Add(1)go func() {defer g.done()if err := f(); err != nil {// 记录首次goroutine返回的errg.errOnce.Do(func() {g.err = errif g.cancel != nil {// 传递取消信号g.cancel(g.err)}})}}()
}

done

控制活跃goroutine数量的一环

func (g *Group) done() {if g.sem != nil {<-g.sem}g.wg.Done()
}

WithContext

根据传入的context，返回一个派生的context和一个有context的group

派生的context会在传递给GO()/TryGo()的函数首次返回非nil错误或Wait()首次返回时被取消，以先发生者为主

func WithContext(ctx context.Context) (*Group, context.Context) {ctx, cancel := withCancelCause(ctx)return &Group{cancel: cancel}, ctx
}

Wait

阻塞直到

func (g *Group) Wait() error {g.wg.Wait()if g.cancel != nil {g.cancel(g.err)}return g.err
}

TryGo

非阻塞的Go方法

如果调用了SetLimit()，调用Go()方法会阻塞

而TryGo()不会阻塞，如果因为goroutine数量限制未能调用函数就会返回false

成功调用就会返回true

func (g *Group) TryGo(f func() error) bool {if g.sem != nil {// 非阻塞式检测g.sem是否还有容量select {case g.sem <- token{}:default:return false}}g.wg.Add(1)go func() {defer g.done()if err := f(); err != nil {g.errOnce.Do(func() {g.err = errif g.cancel != nil {g.cancel(g.err)}})}}()return true
}

参考

https://jianghushinian.cn/2024/11/04/x-sync-errgroup/