Go并发编程中sync/errGroup的使用
作者:深度思维者 发布时间:2024-02-05 04:40:56
目录
一.序
二.errGroup
2.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
来源:https://www.cnblogs.com/failymao/p/15677782.html
猜你喜欢
- 导语一直以来,中国象棋都是中华民族的一种象征,当然也是人们最为喜感的一种娱乐方式。在若干年前,人们都习惯于约上自己的棋友,来一种激战。可是,
- OpenCV转换成PIL.Image格式:import cv2 from PIL import Image import numpy img
- 最近在研究Hacker News API时遇到一个HTTPS问题。因为所有的Hacker News API都是通过加密的HTTPS协议访问的
- 一、利用webbrowser.open()打开一个网站:>>> import webbrowser >>>
- 引言最近在技术交流群里聊到一个关于图像文字识别的需求,在工作、生活中常常会用到,比如票据、漫画、扫描件、照片的文本提取。博主基于 PyQt
- 搭建FTP,或者是搭建网络文件系统,这些方法都能够实现Linux的目录共享。但是FTP和网络文件系统的功能都过于强大,因此它们都有一些不够方
- 在本身比较复杂的页面里,再突出信息,往往是把几种方法叠加起来使用,比如加粗加大、加粗加色等,区别在于使用的类别和程度。导致的结果是呈现越来越
- 现在对于python2版本,直接修改python.exe名字为python2.exe,命令python2 -m pip install --
- 对Python中正则表达式的理解,主要就是对符号的理解,本文即对Python中常用的正则表达式符号进行简析。其主要的符号有:.默认匹配一个字
- 1.1.1 摘要 Join是关系型数据库系统的重要操作之一,SQL Server中包含的常用Join:内联接、外联接和交叉联接等。如果我们想
- MySQL是关系型数据库中的明星,MongoDB是文档型数据库中的翘楚。下面通过一个设计实例对比一下二者:假设我们正在维护一个手机产品库,里
- import osfrom PIL import Image#批量剪切目录下图片for j in range(10,121): &
- 以下代码已经在SQLServer2008上的示例数据库测试通过问题一:如何为数据进行加密与解密,避免使用者窃取机密数据? 对于一些敏感数据,
- Jupyter Notebook是一个基于浏览器的交互式编程环境(REPL, read eval print loop),它主要构建在IPy
- 很久以前就知道微软的Petshop的很经典,昨天抽出时间去学习,一开始还真的不适应,什么成员资格,还真的看不太懂,运行petshop想从登陆
- Python自动化办公之删除重复的视频文件获取所有文件路径import osimport filecmpfile_list=[]new_li
- Python解释器的相关配置,供大家参考,具体内容如下1、准备工作(1)Pycharm版本为3.4或者更高。(2)电脑上至少已经安装了一个P
- 这个是作者自己封装的一个钉钉机器人的包,目前只支持发文本格式、链接格式、markdown格式的消息,我们可以在很多场景用到这个,比如告警通知
- 阿里云服务器的带宽为2M,网站每日的备份包都3G多了,离线下载太费时间了,打算每日将备份包自动上传到自己的百度云盘里。 1、先安装
- 一、统计语句1、--统计当前【>当天00点以后的数据】SELECT * FROM 表 WHERE CONVERT(Nvarchar,