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

 更新时间:2022-01-27 14:15:52   作者:佚名   我要评论(0)

目录一.序二.errGroup2.1 函数签名
三.源码3.1 Group
3.2 WaitContext
3.3 Go
3.4 Wait
四. 案例五. 参考一.序


这一篇算是并发编程的

一.序

这一篇算是并发编程的一个补充,起因是当前有个项目,大概の 需求是,根据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 <- struct{}{}
 })
 
 server := http.Server{
  Handler: mux,
  Addr:    ":8099",
 }
 
 // g1
 // g1 退出了所有的协程都能退出么?
 // g1 退出后, context 将不再阻塞,g2, g3 都会随之退出
 // 然后 main 函数中的 g.Wait() 退出,所有协程都会退出
 g.Go(func() error {
  err := server.ListenAndServe() // 服务启动后会阻塞, 虽然使用的是 go 启动,但是由于 g.WaitGroup 试得其是个阻塞的 协程
  if err != nil {
   log.Println("g1 error,will exit.", err.Error())
  }
  return err
 })
 
 // g2
 // g2 退出了所有的协程都能退出么?
 // 到调用 `/shutdown`接口时, serverOut 无缓冲管道写入数据, case接收到数据后执行server.shutdown, 此时 g1 httpServer会退出
 // g1退出后,会返回error,将error加到g中,同时会调用 cancel()
 // g3 中会 select case ctx.Done, context 将不再阻塞,g3 会随之退出
 // 然后 main 函数中的 g.Wait() 退出,所有协程都会退出
 g.Go(func() error {
  select {
  case <-ctx.Done():
   log.Println("g2 errgroup exit...")
  case <-serverOut:
   log.Println("g2, request `/shutdown`, server will out...")
  }
 
  timeoutCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
  // 这里不是必须的,但是如果使用 _ 的话静态扫描工具会报错,加上也无伤大雅
  defer cancel()
 
  err := server.Shutdown(timeoutCtx)
  log.Println("shutting down server...")
  return err
 })
 
 // g3
 // g3 捕获到 os 退出信号将会退出
 // g3 退出了所有的协程都能退出么?
 // g3 退出后, context 将不再阻塞,g2 会随之退出
 // g2 退出时,调用了 shutdown,g1 会退出
 // 然后 main 函数中的 g.Wait() 退出,所有协程都会退出
 g.Go(func() error {
  quit := make(chan os.Signal, 0)
  signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
 
  select {
  case <-ctx.Done():
   log.Println("g3, ctx execute cancel...")
   log.Println("g3 error,", ctx.Err().Error())
   // 当g2退出时,已经有错误了,此时的error 并不会覆盖到g中
   return ctx.Err()
  case sig := <-quit:
   return fmt.Errorf("g3 get os signal: %v", sig)
  }
 })
 
 // g.Wait 等待所有 go执行完毕后执行
 fmt.Printf("end, errgroup exiting, %+v\n", g.Wait())
}

运行测试

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的使用的文章就介绍到这了,更多相关Go sync/errGroup内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
  • Go并发编程之sync.Once使用实例详解
  • Go语言并发编程 sync.Once
  • Go 并发读写 sync.map 详细
  • Go语言中sync.Cond使用详解
  • Golang之sync.Pool使用详解
  • 在golang中使用Sync.WaitGroup解决等待的问题
  • Golang中的sync包的WaitGroup操作
  • 深入Golang中的sync.Pool详解
  • Django如何使用asyncio协程和ThreadPoolExecutor多线程
  • golang中使用sync.Map的方法

相关文章

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

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

    目录一.序二.errGroup2.1 函数签名 三.源码3.1 Group 3.2 WaitContext 3.3 Go 3.4 Wait 四. 案例五. 参考一.序 这一篇算是并发编程的
    2022-01-27
  • Go语言原子操作及互斥锁的区别

    Go语言原子操作及互斥锁的区别

    目录增或减比较并交换(Compare And Swap)载入与存储交换原子值原子操作与互斥锁的区别原子操作就是不可中断的操作,外界是看不到原子操作的
    2022-01-27
  • Go语言并发爬虫的具体实现

    Go语言并发爬虫的具体实现

    目录写在前面1. 单线程爬虫2. 多线程爬虫2.1 channel main函数2.2 sync.WaitGroup3. 源码地址写在前面 这篇文章主要让大家明白多线程爬虫,
    2022-01-27
  • Go?container包的介绍

    Go?container包的介绍

    目录1.简介2.list2.1数据结构 2.2插入元素3.ring3.1数据结构 4.heap4.1数据结构 1.简介 Container — 容器数据类型:该包实现了三个复
    2022-01-27
  • Go语言做爬虫状态码返回418的问题解决

    Go语言做爬虫状态码返回418的问题解决

    目录背景原因分析代码部分背景 ?在使用Go语言做爬虫时,使用http.Get(url)去获取网页内容,状态码返回404,Body体为空。 原因分析 ?http
    2022-01-27
  • golang实现数组分割的示例代码

    golang实现数组分割的示例代码

    需求:给定一个数组和一个正整数,要求把数组分割成多个正整数大小的数组,如果不够分,则最后一个数组分到剩余的所有元素。 示例1:
    2022-01-27
  • Go语言参数传递是传值还是传引用

    Go语言参数传递是传值还是传引用

    目录什么是传值(值传递)什么是传引用(引用传递)迷惑Mapchan类型和map、chan都不一样的slice小结对于了解一门语言来说,会关心我们在函数调
    2022-01-27
  • vscode搭建go开发环境案例详解

    vscode搭建go开发环境案例详解

    目录前言一、安装goLang二.配置环境变量三、vscode安装插件四.安装golang依赖五.新建go文件提示:文章写完后,目录可以自动生成,如何生成可
    2022-01-27
  • 浅谈Go1.18中的泛型编程

    浅谈Go1.18中的泛型编程

    目录前言以前的Go泛型泛型是什么Go的泛型泛型函数泛型类型类型集合和接口的差异总结前言 经过这几年的千呼万唤,简洁的Go语言终于在1.18版
    2022-01-27
  • golang强制类型转换和类型断言

    golang强制类型转换和类型断言

    目录1.强制类型转换2.类型断言1.强制类型转换 golang是强类型语言,但是和c++,Java不太一样的是,go中没有隐式类型转换,go中的类型转换只有
    2022-01-27

最新评论