首页 > 作文

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

更新时间:2023-04-04 02:27:53 阅读: 评论:0

目录
一.序 二.errgroup 2.1 函数签名 三.源码 3.1 group 3.2 waitcontext 3.3 go 3.4 wait 四. 案例 五. 参考

一.序

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

问题其实很明确: 如果在初始化分区消费者时,只要有一个消费创建失败,那么初始化工作就算失败,程序应该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 可以创建一个带取消的groupgo 方法传入一个 func() error 内部会启动一个goroutine 去处理wait 类似waitgroup的wait 方法,等待所有的 goroutine结束后退出,返回的错误是一个出错的 err

三.源码

3.1 group

type group struct {    // context 的 cancel 方法 cancel func()     // 复用 waitgroup wg sync.wa奖励办法itgroup  // 用来保证只会接受一次错误 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 rver 的启动和关闭 ,以及 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.newrvemux() mux.handlefunc("/ping", func(w http.responwriter, r *http.request) {  _, _ = w.write([]byte("pong")) })  // 模拟单个服务错误退出 rverout := make(chan struct{}) mux.handlefunc("/shutdown", func(w http.responwriter, r *http.request) {  rverout <- struct{}{} })  rver := http.rver{  handler: mux,  addr:    ":8099", }  // g1 // g1 退出了所有的协程都能退出么? // g1 退出后, context 将不再阻塞,g2, g3 都会随之退出 // 然后 main 函数中的 g.wait() 退出,所有协程都会退出 g.go(func() error {  err := rver.listenandrve() // 服务启动后会阻塞, 虽然使用的是 go 启动,但是由于 g.w友情短信大全aitgroup 试得其是个阻塞的 协程  if err != nil {   log.println("g1 error,will exit.", err.error())  }  return err })  // g2 // g2 退出了所有的协程都能退出么? // 到调用 `/shutdown`接口时, rverout 无缓冲管道写入数据, ca接收到数据后执行rver.shutdown, 此时 g1 httprver会退出 // g1退出后,会返回error,将error加到g中,同时会调用 cancel() // g3 中会 lect ca ctx.done, context 将不再阻塞,g3 会随之退出 // 然后 main 函数中的 g.wait() 退出,所有协程都会退出 g.go(func() error {  lect {  ca <-ctx.done():   log.println("g2 errgroup exit...")  ca <-rverout:   log.println("g2, request `/shutdown`, rver will out...")  }   timeoutctx, cancel := context.withtimeout(context.background(), 3*time.cond)  // 这里不是必须的,但是如果使用 _ 的话静态扫描工具会报错,加上也无伤大雅  defer cancel()   err := rver.shutdown(timeoutctx)  log.println("shutting down rver...")  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)   lect {  ca <-ctx.done():   log.println("g3, ctx execute cancel...")   log.println("g3 error,", ctx.err().error())   // 当g2退出时,已经有错误了,此时的error 并不会覆盖到g中   return ctx.err()  ca 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_ba\daily_test\errorgroup>go run demo.go

浏览器输入http://127.0.0.1:8099/shutdown

控制台输出

2021/12/11 10:52:03 g2, request `/shutdown`, rver will out…
2021/12/11 10:52:03 g1 error,will ex星期用英语怎么说it. http: rver clod
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 rver…
end, errgroup exiting, http: rver clod

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

在terminal 按ctrl + c

输出
2021/12/11 10:55:51 g2 errgroup exit…
2021/12/11 10:55:51 g1 error,will exit. http: rver clod
2021/12/11 10:55:51 shutting down rver…
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内容请搜索www.887551.com以前的文章或继续浏览下面的相关文章希望大家以后多多支持www.887551.com!

本文发布于:2023-04-04 02:27:49,感谢您对本站的认可!

本文链接:https://www.wtabcd.cn/fanwen/zuowen/17dc089a18f8128eb9a80a576207e1cd.html

版权声明:本站内容均来自互联网,仅供演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,我们将在24小时内删除。

本文word下载地址:Go并发编程中sync/errGroup的使用.doc

本文 PDF 下载地址:Go并发编程中sync/errGroup的使用.pdf

下一篇:返回列表
标签:分区   消费者   都会   数据
相关文章
留言与评论(共有 0 条评论)
   
验证码:
Copyright ©2019-2022 Comsenz Inc.Powered by © 专利检索| 网站地图