我们把一个复杂的任务，尤其是依赖多个微服务 rpc 需要聚合数据的任务，分解为依赖和并行，依赖的意思为: 需要上游 a 的数据才能访问下游 b 的数据进行组合。但是并行的意思为: 分解为多个小任务并行执行，最终等全部执行完毕。

https://pkg.go.dev/golang.org/x/sync/errgroup 核心原理: 利用 sync.Waitgroup 管理并行执行的 goroutine。 3/10.go

并行工作流

错误处理 或者 优雅降级

context 传播和取消

利用局部变量+闭包

为什么要有sync.errgroup
go支持并发，一般采用的是 channel 、 sync.WaitGroup 、context，来实现各个协程之间的流程控制和消息传递。
但是对于开启的成千上万的协程，如果在每个协程内都自行去打印 错误日志的话，会造成日志分散，不好分析。
所以我们要实现一种能统一处理各个协程错误的工具

什么是 sync.errgroup
Go团队在实验仓库中添加了一个名为sync.errgroup的新软件包。 sync.ErrGroup在sync.WaitGroup功能的基础上，增加了错误传递，以及在发生不可恢复的错误时取消整个goroutine集合，或者等待超时

主要是利用了 waitgroup,context以及sync.Once，对这三个不熟悉的应先去看下相应的知识点

获取方法

go get golang.org/x/sync

errgroup 的功能

1、处理子协程 error

func main() {    var g errgroup.Group  // 声明一个group实例    var urls = []string{        "http://www.golang.org/",        "http://www.google.com/",        "http://www.somestupidname.com/",    }    for _, url := range urls {  // 分别获取网站内容        url := url // url是局部变量，for循环中对多个协程传递值时，需要重新进行赋值        g.Go(func() error {  // group 的go方法，启一个协程去执行代码            // Fetch the URL.            resp, err := http.Get(url)            if err == nil {                resp.Body.Close()            }            return err        })    }    if err := g.Wait(); err == nil {  // group 的wait方法，等待上面的 g.go的协程执行完成，并且可以接受错误        fmt.Println("Successfully fetched all URLs.")    }}

上面这个例子是简单的利用 errgroup 进行的 waitGroup和error的处理，下面我们对关键的代码做一个分析，并结合源码来看

var g errgroup.Group

声明一个 group的实例，我们看下 group 包含哪些东西

type Group struct {    cancel func()    wg sync.WaitGroup    errOnce sync.Once    err     error}

group是一个结构体，包含四个部分

• cancel 一个取消的函数，主要来包装context.WithCancel的CancelFunc
• wg 借助于WaitGroup实现的
• errOnce 使用sync.Once实现只输出第一个err
• err 记录下错误的信息

g.Go(func() error {}

启动goroutine 执行代码
记录第一个出错的goroutine的err信息。我们看下源码

func (g *Group) Go(f func() error) {    g.wg.Add(1)  // 和WaitGroup 一样，每执行一个新的g，通过add方法 加1    go func() {        defer g.wg.Done() // 执行结束后 调用 Done方法，减1        if err := f(); err != nil {  // 执行传入的匿名函数            g.errOnce.Do(func() {   // 如果匿名函数返回错误，会记录错误信息。注意这里用的 once.Do，只执行一次，仅会记录第一个出现的err                g.err = err                if g.cancel != nil {  // 如果初始化的有 cancel 函数，会调用 cancel退出                    g.cancel()                }            })        }    }()}

再来看下 g.Wait()

func (g *Group) Wait() error {    g.wg.Wait()  // 和 WaitGroup 一样，在主线程调用 wait 方法，阻塞等待所有g执行完成    if g.cancel != nil {  // 如果初始化了 cancel 函数，就执行        g.cancel()    }    return g.err  // 返回第一个出现的err信息}

2、结合 context 来使用

package mainimport (    "context"    "fmt"    "golang.org/x/sync/errgroup"    "time")func main() {    ctx, cancel := context.WithCancel(context.Background())    group, errCtx := errgroup.WithContext(ctx)    for index := 0; index < 3; index++ {        indexTemp := index        // 新建子协程        group.Go(func() error {            fmt.Printf("indexTemp=%d \n", indexTemp)            if indexTemp == 0 { // 第一个协程                fmt.Println("indexTemp == 0 start ")                fmt.Println("indexTemp == 0 end")            } else if indexTemp == 1 { // 第二个协程                fmt.Println("indexTemp == 1 start")                //这里一般都是某个协程发生异常之后，调用cancel()                //这样别的协程就可以通过errCtx获取到err信息，以便决定是否需要取消后续操作                cancel() // 第二个协程异常退出                fmt.Println("indexTemp == 1 err ")            } else if indexTemp == 2 {                fmt.Println("indexTemp == 2 begin")                // 休眠1秒，用于捕获子协程2的出错                time.Sleep(1 * time.Second)                //检查 其他协程已经发生错误，如果已经发生异常，则不再执行下面的代码                err := CheckGoroutineErr(errCtx) // 第三个协程感知第二个协程是否正常                if err != nil {                    return err                }                fmt.Println("indexTemp == 2 end ")            }            return nil        })    }    // 捕获err    err := group.Wait()    if err == nil {        fmt.Println("都完成了")    } else {        fmt.Printf("get error:%v", err)    }}//校验是否有协程已发生错误func CheckGoroutineErr(errContext context.Context) error {    select {    case <-errContext.Done():        return errContext.Err()    default:        return nil    }}

本文来自博客园，作者：孙龙-程序员，转载请注明原文链接：
https://www.cnblogs.com/sunlong88/p/16118676.html