Golang errgroup 设计及实现原理解析

开篇

errgroup 源码拆解

Group

WithContext

Wait

SetLimit

TryGo

使用方法

结束语

开篇

继上次学习了信号量 semaphore 扩展库的设计思路和实现之后，今天我们继续来看 golang.org/x/sync 包下的另一个经常被 Golang 开发者使用的大杀器：errgroup。

业务研发中我们经常会遇到需要调用多个下游的场景，比如加载一个商品的详情页，你可能需要访问商品服务，库存服务，券服务，用户服务等，才能从各个数据源获取到所需要的信息，经过一些鉴权逻辑后，组装成前端需要的数据格式下发。

串行调用当然可以，但这样就潜在的给各个调用预设了【顺序】，必须执行完 A,B,C 之后才能执行 D 操作。但如果我们对于顺序并没有强需求，从语义上看两个调用是完全独立可并发的，那么我们就可以让他们并发执行。

这个时候就可以使用 errgroup 来解决问题。一定意义上讲，errgroup 是基于 WaitGroup 在错误传递上进行一些优化而提供出来的能力。它不仅可以支持 context.Context 的相关控制能力，还可以将子任务的 error 向上传递。

errgroup 源码拆解

errgroup 定义在 golang.org/x/sync/errgroup，承载核心能力的结构体是 Group。

Group

type token struct{}
// A Group is a collection of goroutines working on subtasks that are part of
// the same overall task.
//
// A zero Group is valid, has no limit on the number of active goroutines,
// and does not cancel on error.
type Group struct {
cancel func()
wg sync.WaitGroup
sem chan token
errOnce sync.Once
err     error
}

Group 就是对我们上面提到的一堆子任务执行计划的抽象。每一个子任务都会有自己对应的 goroutine 来执行。

通过这个结构我们也可以看出来，errgroup 底层实现多个 goroutine 调度，等待的能力还是基于 sync.WaitGroup。

WithContext

我们可以调用 errgroup.WithContext 函数来创建一个 Group。

// WithContext returns a new Group and an associated Context derived from ctx.
//
// The derived Context is canceled the first time a function passed to Go
// returns a non-nil error or the first time Wait returns, whichever occurs
// first.
func WithContext(ctx context.Context) (*Group, context.Context) {
ctx, cancel := context.WithCancel(ctx)
return &amp;Group{cancel: cancel}, ctx
}

这里可以看到，Group 的 cancel 函数本质就是为了支持 context 的 cancel 能力，初始化的 Group 只有一个 cancel 属性，其他都是默认的。一旦有一个子任务返回错误，或者是 Wait 调用返回，这个新 Context 就会被 cancel。

Wait

本质上和 WaitGroup 的 Wait 方法语义一样，既然我们是个 group task，就需要等待所有任务都执行完，这个语义就由 Wait 方法提供。如果有多个子任务返回错误，它只会返回第一个出现的错误，如果所有的子任务都执行成功，就返回 nil。


// Wait blocks until all function calls from the Go method have returned, then
// returns the first non-nil error (if any) from them.
func (g *Group) Wait() error {
g.wg.Wait()
if g.cancel != nil {
g.cancel()
}
return g.err
}

Wait 的实现非常简单。一个前置的 WaitGroup Wait，结束后只做了两件事：

如果对于公共的 Context 有 cancel 函数，就将其 cancel，因为事情办完了；

返回公共的 Group 结构中的 err 给调用方。

Group 的核心能力就在于能够并发执行多个子任务，从调用者的角度，我们只需要传入要执行的函数，签名为：func() error即可，非常通用。如果任务执行成功，就返回 nil，否则就返回 error，并且会 cancel 那个新的 Context。底层的调度逻辑由 Group 的 Go 方法实现：

// Go calls the given function in a new goroutine.
// It blocks until the new goroutine can be added without the number of
// active goroutines in the group exceeding the configured limit.
//
// The first call to return a non-nil error cancels the group; its error will be
// returned by Wait.
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 {
g.errOnce.Do(func() {
g.err = err
if g.cancel != nil {
g.cancel()
}
})
}
}()
}
func (g *Group) done() {
if g.sem != nil {
<-g.sem
}
g.wg.Done()
}

我们重点来分析下 Go 这里发生了什么。

WaitGroup 加 1 用作计数；

启动一个新的 goroutine 执行调用方传入的 f() 函数；

若 err 为 nil 说明执行正常；

若 err 不为 nil，说明执行出错，此时将这个返回的 err 赋值给全局 Group 的变量 err，并将 context cancel 掉。注意，这里的处理在 once 分支中，也就是只有第一个来的错误会被处理。

在 defer 语句中调用 Group 的 done 方法，底层依赖 WaitGroup 的 Done，说明这一个子任务结束。

这里也可以看到，其实所谓 errgroup，我们并不是将所有子任务的 error 拼成一个字符串返回，而是直接在 Go 方法那里将第一个错误返回，底层依赖了 once 的能力。

SetLimit

其实看到这里，你有没有觉得 errgroup 的功能有点鸡肋？底层核心技术都是靠 WaitGroup 完成的，自己只不过是起了个 goroutine 执行方法，err 还只能保留一个。而且 Group 里面的 sem 那个 chan 是用来干什么呢？

这一节我们就来看看，Golang 对 errgroup 能力的一次扩充。

到目前为止，errgroup 是可以做到一开始人们对它的期望的，即并发执行子任务。但问题在于，这里是每一个子任务都开了个goroutine，如果是在高并发的环境里，频繁创建大量goroutine 这样的用法很容易对资源负载产生影响。开发者们提出，希望有办法限制 errgroup 创建的 goroutine 数量，参照这个 proposal: #27837

// SetLimit limits the number of active goroutines in this group to at most n.
// A negative value indicates no limit.
//
// Any subsequent call to the Go method will block until it can add an active
// goroutine without exceeding the configured limit.
//
// The limit must not be modified while any goroutines in the group are active.
func (g *Group) SetLimit(n int) {
if n < 0 {
g.sem = nil
return
}
if 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)
}

SetLimit 的参数 n 就是我们对这个 Group 期望的最大 goroutine 数量，这里其实除去校验逻辑，只干了一件事：g.sem = make(chan token, n)，即创建一个长度为 n 的 channel，赋值给 sem。

回忆一下我们在 Go 方法 defer 调用 done 中的表现，是不是清晰了很多？我们来理一下：

首先，Group 结构体就包含了 sem 变量，只作为通信，元素是空结构体，不包含实际意义：

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

如果你对整个 Group 的 Limit 没有要求，which is fine，你就直接忽略这个 SetLimit，errgroup 的原有能力就可以支持你的诉求。

但是，如果你希望保持 errgroup 的 goroutine 在一个上限之内，请在调用 Go 方法前，先 SetLimit，这样 Group 的 sem 就赋值为一个长度为 n 的 channel。

那么，当你调用 Go 方法时，注意下面的框架代码，此时 g.sem 不为 nil，所以我们会塞一个 token 进去，作为占坑，语义上表示我申请一个 goroutine 用。

func (g *Group) Go(f func() error) {
if g.sem != nil {
g.sem <- token{}
}
g.wg.Add(1)
go func() {
defer g.done()
                ...
}()
}

当然，如果此时 goroutine 数量已经达到上限，这里就会 block 住，直到别的 goroutine 干完活，sem 输出了一个 token之后，才能继续往里面塞。

在每个 goroutine 执行完毕后 defer 的 g.done 方法，则是完成了这一点：

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

这样就把 sem 的用法串起来了。我们通过创建一个定长的channel来实现对于 goroutine 数量的管控，对于channel实际包含的元素并不关心，所以用一个空结构体省一省空间，这是非常优秀的设计，大家平常也可以参考。

TryGo

TryGo 和 SetLimit 这套体系其实都是不久前欧长坤大佬提交到 errgroup 的能力。

一如既往，所有带 TryXXX的函数，都不会阻塞。其实办的事情非常简单，和 Go 方法一样，传进来一个 func() error来执行。

和 Go 方法的区别在于，如果判断 limit 已经不够了，此时不再阻塞，而是直接 return false，代表执行失败。其他的部分完全一样。

// TryGo calls the given function in a new goroutine only if the number of
// active goroutines in the group is currently below the configured limit.
//
// The return value reports whether the goroutine was started.
func (g *Group) TryGo(f func() error) bool {
if g.sem != nil {
select {
case g.sem <- token{}:
// Note: this allows barging iff channels in general allow barging.
default:
return false
}
}
g.wg.Add(1)
go func() {
defer g.done()
if err := f(); err != nil {
g.errOnce.Do(func() {
g.err = err
if g.cancel != nil {
g.cancel()
}
})
}
}()
return true
}

使用方法

这里我们先看一个最常见的用法，针对一组任务，每一个都单独起 goroutine 执行，最后获取 Wait 返回的 err 作为整个 Group 的 err。


package main
import (
    "errors"
    "fmt"
    "time"
    "golang.org/x/sync/errgroup"
)
func main() {
    var g errgroup.Group
    // 启动第一个子任务,它执行成功
    g.Go(func() error {
        time.Sleep(5 * time.Second)
        fmt.Println("exec #1")
        return nil
    })
    // 启动第二个子任务，它执行失败
    g.Go(func() error {
        time.Sleep(10 * time.Second)
        fmt.Println("exec #2")
        return errors.New("failed to exec #2")
    })
    // 启动第三个子任务，它执行成功
    g.Go(func() error {
        time.Sleep(15 * time.Second)
        fmt.Println("exec #3")
        return nil
    })
    // 等待三个任务都完成
    if err := g.Wait(); err == nil {
        fmt.Println("Successfully exec all")
    } else {
        fmt.Println("failed:", err)
    }
}

你会发现，最后 err 打印出来就是第二个子任务的 err。

当然，上面这个 case 是我们正好只有一个报错，但是，如果实际有多个任务都挂了呢？

从完备性来考虑，有没有什么办法能够将多个任务的错误都返回呢？

这一点其实 errgroup 库并没有提供非常好的支持，需要开发者自行做一些改造。因为 Group 中只有一个 err 变量，我们不可能基于 Group 来实现这一点。

通常情况下，我们会创建一个 slice 来存储 f() 执行的 err。


package main
import (
    "errors"
    "fmt"
    "time"
    "golang.org/x/sync/errgroup"
)
func main() {
    var g errgroup.Group
    var result = make([]error, 3)
    // 启动第一个子任务,它执行成功
    g.Go(func() error {
        time.Sleep(5 * time.Second)
        fmt.Println("exec #1")
        result[0] = nil // 保存成功或者失败的结果
        return nil
    })
    // 启动第二个子任务，它执行失败
    g.Go(func() error {
        time.Sleep(10 * time.Second)
        fmt.Println("exec #2")
        result[1] = errors.New("failed to exec #2") // 保存成功或者失败的结果
        return result[1]
    })
    // 启动第三个子任务，它执行成功
    g.Go(func() error {
        time.Sleep(15 * time.Second)
        fmt.Println("exec #3")
        result[2] = nil // 保存成功或者失败的结果
        return nil
    })
    if err := g.Wait(); err == nil {
        fmt.Printf("Successfully exec all. result: %v\n", result)
    } else {
        fmt.Printf("failed: %v\n", result)
    }
}

可以看到，我们声明了一个 result slice，长度为 3。这里不用担心并发问题，因为每个 goroutine 读写的位置是确定唯一的。

本质上可以理解为，我们把 f() 返回的 err 不仅仅给了 Group 一份，还自己存了一份，当出错的时候，Wait 返回的错误我们不一定真的用，而是直接用自己错的这一个 error slice。

Go 官方文档中的利用 errgroup 实现 pipeline 的示例也很有参考意义：由一个子任务遍历文件夹下的文件，然后把遍历出的文件交给 20 个 goroutine，让这些 goroutine 并行计算文件的 md5。

这里贴出来简化代码学习一下.

package main
import (
"context"
"crypto/md5"
"fmt"
"io/ioutil"
"log"
"os"
"path/filepath"
"golang.org/x/sync/errgroup"
)
// Pipeline demonstrates the use of a Group to implement a multi-stage
// pipeline: a version of the MD5All function with bounded parallelism from
// https://blog.golang.org/pipelines.
func main() {
m, err := MD5All(context.Background(), ".")
if err != nil {
log.Fatal(err)
}
for k, sum := range m {
fmt.Printf("%s:\t%x\n", k, sum)
}
}
type result struct {
path string
sum  [md5.Size]byte
}
// MD5All reads all the files in the file tree rooted at root and returns a map
// from file path to the MD5 sum of the file's contents. If the directory walk
// fails or any read operation fails, MD5All returns an error.
func MD5All(ctx context.Context, root string) (map[string][md5.Size]byte, error) {
// ctx is canceled when g.Wait() returns. When this version of MD5All returns
// - even in case of error! - we know that all of the goroutines have finished
// and the memory they were using can be garbage-collected.
g, ctx := errgroup.WithContext(ctx)
paths := make(chan string)
g.Go(func() error {
defer close(paths)
return filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if !info.Mode().IsRegular() {
return nil
}
select {
case paths <- path:
case <-ctx.Done():
return ctx.Err()
}
return nil
})
})
// Start a fixed number of goroutines to read and digest files.
c := make(chan result)
const numDigesters = 20
for i := 0; i < numDigesters; i++ {
g.Go(func() error {
for path := range paths {
data, err := ioutil.ReadFile(path)
if err != nil {
return err
}
select {
case c <- result{path, md5.Sum(data)}:
case <-ctx.Done():
return ctx.Err()
}
}
return nil
})
}
go func() {
g.Wait()
close(c)
}()
m := make(map[string][md5.Size]byte)
for r := range c {
m[r.path] = r.sum
}
// Check whether any of the goroutines failed. Since g is accumulating the
// errors, we don't need to send them (or check for them) in the individual
// results sent on the channel.
if err := g.Wait(); err != nil {
return nil, err
}
return m, nil
}

其实本质上还是 channel发挥了至关重要的作用，这里建议大家有时间尽量看看源文章：pkg.go.dev/golang.org/…

对于用 errgroup 实现 pipeline 模式有很大帮助。

结束语

今天我们学习了 errgroup 的源码已经新增的 SetLimit 能力，其实看过了 sync 相关的这些库，整体用到的能力其实大差不差，很多设计思想都是非常精巧的。比如 errgroup 中利用定长 channel 实现控制 goroutine 数量，空结构体节省内存空间。

并且 sync 包的实现一般都还是非常简洁的，比如 once，singleflight，semaphore 等。建议大家有空的话自己过一遍，对并发和设计模式的理解会更上一个台阶。

errgroup 本身并不复杂，业界也有很多封装实现，大家可以对照源码再思考一下还有什么可以改进的地方。

以上就是Golang errgroup 设计及实现原理解析的详细内容，更多关于Golang errgroup 设计原理的资料请关注易知道(ezd.cc)其它相关文章！

推荐阅读