Go并发编程中sync/errGroup的使用

本篇文章给大家分享《Go并发编程中sync/errGroup的使用》，覆盖了Golang的常见基础知识，其实一个语言的全部知识点一篇文章是不可能说完的，但希望通过这些问题，让读者对自己的掌握程度有一定的认识(B 数)，从而弥补自己的不足，更好的掌握它。

一.序

这一篇算是并发编程的一个补充，起因是当前有个项目，大概の 需求是，根据kafka的分区（partition）数，创建同等数量的 消费者（ goroutine）从不同的分区中消费者消费数据，但是总有某种原因导致，某一个分区消费者创建失败，但是其他分区消费者创建失败。 最初的逻辑是，忽略分区失败的逻辑，将成功创建的分区消费者收集，用于获取消息进行数据处理。 代码就不在这里展示。

问题其实很明确： 如果在初始化分区消费者时，只要有一个消费创建失败，那么初始化工作就算失败，程序应该panic，退出。

但是当初设计时，消费者负责从kafka上游的某个topic获取到数据，然后经过数据处理后，再通过生产者将处理后的数据发送到下游的topic中，由于当时设计时，代码耦合比较重，导致无法通过初始化工作做这些，只能在启动生产者后， 再创建消费者，这就导致 创建消费者-->获取数据-->处理数据 杂糅到了一起。 这个问题一直到最近才有时间想着来解决。

比如有三个分区创建了三个分区的消费者，每个分区的消费者对应从自己的分区重获取数据，三个分区最初使用waitGroup进行控制三个分区创建，只有当三个分区都创建完成后才会执行后续逻辑。 但是 waitgroup并不能很好的解决：只要一个 goroutine 出错我们就不再等其他 goroutine 了，就默认创建分区消费者失败了，所以此时便想到了 errGroup

二.errGroup

errGroup 是google开源的基础扩展库。使用时先进行下载

go get -u golang.org/x/sync

2.1 函数签名

type Group struct {
 // contains filtered or unexported fields
}
 
    func WithContext(ctx context.Context) (*Group, context.Context)
    func (g *Group) Go(f func() error)
    func (g *Group) Wait() error

整个包就一个 Group 结构体

• 通过WaitContext 可以创建一个带取消的group
• Go 方法传入一个 func() error 内部会启动一个goroutine 去处理
• Wait 类似WaitGroup的Wait 方法，等待所有的 goroutine结束后退出，返回的错误是一个出错的 err

三.源码

3.1 Group

type Group struct {
    // context 的 cancel 方法
 cancel func()
 
    // 复用 WaitGroup
 wg sync.WaitGroup
 
 // 用来保证只会接受一次错误
 errOnce sync.Once
    // 保存第一个返回的错误
 err     error
}

3.2 WaitContext

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

WithContext 就是使用 WithCancel 创建一个可以取消的 context 将 cancel 赋值给 Group 保存起来，然后再将 context 返回回去

注意这里有一个坑，在后面的代码中不要把这个 ctx 当做父 context 又传给下游，因为 errgroup 取消了，这个 context 就没用了，会导致下游复用的时候出错

3.3 Go

func (g *Group) Go(f func() error) {
 g.wg.Add(1)
 
 go func() {
  defer g.wg.Done()
        
        // 通过执行传入的匿名函数返回的错误值判断是否需要执行cancel
  if err := f(); err != nil {
      // 这一点很重要，确保错误只会被执行一次
   g.errOnce.Do(func() {
    g.err = err
    if g.cancel != nil {
     g.cancel()
    }
   })
  }
 }()
}

Go 方法是一个封装，相当于go 关键字的加强，会启动一个携程，然后利用waitgroup 来控制是否结束，如果有一个非 nil 的 error 出现就会保存起来并且如果有 cancel 就会调用 cancel 取消掉，使 ctx 返回

3.4 Wait

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

Wait 方法其实就是调用 WaitGroup 等待，如果有 cancel 就调用一下

四. 案例

基于 errgroup 实现一个 http server 的启动和关闭 ，以及 linux signal 信号的注册和处理，要保证能够 一个退出，全部注销退出。

package main
 
import (
 "context"
 "fmt"
 "log"
 "net/http"
 "os"
 "os/signal"
 "syscall"
 "time"
 
 "golang.org/x/sync/errgroup"
)
 
func main() {
 g, ctx := errgroup.WithContext(context.Background())
 
 mux := http.NewServeMux()
 mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
  _, _ = w.Write([]byte("pong"))
 })
 
 // 模拟单个服务错误退出
 serverOut := make(chan struct{})
 mux.HandleFunc("/shutdown", func(w http.ResponseWriter, r *http.Request) {
  serverOut 

运行测试

D:\gopath\src\Go_base\daily_test\errorGroup>go run demo.go

浏览器输入http://127.0.0.1:8099/shutdown
控制台输出


2021/12/11 10:52:03 g2, request `/shutdown`, server will out...

2021/12/11 10:52:03 g1 error,will exit. http: Server closed

2021/12/11 10:52:03 g3, ctx execute cancel...

2021/12/11 10:52:03 g3 error, context canceled // 间隔了3s

2021/12/11 10:52:06 shutting down server...

end, errgroup exiting, http: Server closed


从执行结果可以看出，这种退出可以保证每个goroutine都能在完成正在执行的工作后退出

在terminal 按ctrl + c

输出

2021/12/11 10:55:51 g2 errgroup exit...

2021/12/11 10:55:51 g1 error,will exit. http: Server closed

2021/12/11 10:55:51 shutting down server...

end, errgroup exiting, g3 get os signal: interrupt


分析都在代码注释中
五. 参考

https://lailin.xyz/post/go-training-week3-errgroup.html
https://pkg.go.dev/golang.org/x/sync/errgroup
到这里，我们也就讲完了《Go并发编程中sync/errGroup的使用》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于golang的知识点！