Go 语言 errgroup 库的使用方式和实现原理

简介: Go 语言 errgroup 库的使用方式和实现原理

介绍

在 Go 语言中,我们可以使用 errgroup 库处理 goroutine 中的错误。

errgroup 库最近更新了,新增支持限制并发数量的功能。

本文我们介绍 errgroup 库的使用方式和实现原理。

使用方式

errgroup 库使用非常简单,我们通过三个简单示例代码,分别介绍三种使用方式。

基础使用

func main() {
 eg := errgroup.Group{}
 eg.Go(func() error {
  fmt.Println("go1")
  return nil
 })
 eg.Go(func() error {
  fmt.Println("go2")
  err := errors.New("go2 err")
  return err
 })
 err := eg.Wait()
 if err != nil {
  fmt.Println("err =", err)
 }
}

阅读上面这段代码,我们使用 errgroup 库的 Go() 方法启动两个 goroutine,分别模拟错误 goroutine 和正常 goroutine

然后,使用 errgroup 库的 Wait() 方法判断是否有 goroutine 返回错误信息。

附加 cancel 功能

func main() {
 eg, ctx := errgroup.WithContext(context.Background())
 eg.Go(func() error {
  time.Sleep(1 * time.Second)
  select {
  case <-ctx.Done():
   fmt.Println("go1 cancel, err = ", ctx.Err())
  default:
   fmt.Println("go1 run")
  }
  return nil
 })
 eg.Go(func() error {
  err := errors.New("go2 err")
  return err
 })
 err := eg.Wait()
 if err != nil {
  fmt.Println("err =", err)
 }
}

阅读上面这段代码,我们使用 errgroup 库的 WithContext() 函数,可以附加 cancel 功能。

我们在第一个使用 Go() 方法启动的协程函数中,使用 select ... case ... default 监听其他协程是否返回错误并做出相应的逻辑处理。

限制并发数量

func main() {
 eg := errgroup.Group{}
 eg.SetLimit(2)
 eg.TryGo(func() error {
  fmt.Println("go1 run")
  return nil
 })
 eg.TryGo(func() error {
  err := errors.New("go2 err")
  return err
 })
 eg.TryGo(func() error {
  fmt.Println("go3 run")
  return nil
 })
 err := eg.Wait()
 if err != nil {
  fmt.Println("err =", err)
 }
}

阅读上面这段代码,我们使用 errgroup 库新增的限制并发数量的功能。

首先,使用 SetLimit() 方法设置并发数量,然后使用 TryGo() 方法替换 Go() 方法。

03

实现原理

我们通过阅读 errgroup 库的源码,简单介绍 errgroup 的实现原理。

我们先阅读 Group 结构体的源码。

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

在源码中,我们可以发现 Group 结构体包含的 5 个字段,其中 sem 字段是最近为了实现限制并发数量功能而新增的。

通过 Group 结构体的字段,我们可以看出 errgroup 实际上是对 synccontext 的封装。

其中,cancel 是使用 contextcancel 方法;wg 是使用 sync.WairGroup 的相关方法;sem 是通过 channel 实现控制并发数量;errOnce 是使用 sync.Once 的特性,只保存第一个返回的 goroutine 错误;errgoroutine 返回的错误。

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()
    }
   })
  }
 }()
}

我们阅读 errgroup 库的 Go() 方法,首先,通过判断 g.sem 的值是否是 nil,如果 g.sem 的值不是 nil,说明已设置并发数量,就通过向 g.sem 中发送一个空结构体 token{},来抢占资源。

如果抢到资源,就启动一个 goroutine,否则,就阻塞,等待其他正在执行的 goroutine 释放一个资源。

细心的读者可能已经发现,Go() 方法除了开头新增判断 g.sem 的值是否为 nil 的逻辑代码之外,defer 也发生了变化,由之前的直接调用 sync.WaitGroupDone() 方法,改为调用 errgroup 库新增的 done() 方法。

done() 方法源码:

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

通过阅读 done() 方法的源码,我们可以发现,在调用 sync.WaitGroupDone() 方法之前,先判断 g.sem 的值是否是 nil,如果不是 nil,则释放资源。

我们再阅读 Wait() 方法的源码:

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

通过阅读 Wait() 方法的源码,我们可以发现它实际上是封装 sync.WaitGroupWait() 方法,和 context 包的 cancel,并且返回所有运行的 goroutine 中第一个返回的错误。

最后,我们阅读新增控制并发数量的功能 TryGo() 方法和 SetLimit() 方法的源码:

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
}

通过阅读 TryGo() 方法的源码,我们可以发现,它和 Go() 方法的区别就是在处理 g.sem 的值上,使用的逻辑不同。

TryGo() 方法在处理 g.sem 的值时,使用 select ... case ... default 语句,先尝试一次抢占资源,当无法抢到资源时,不再阻塞,而是直接返回 false,表示执行失败。

SetLimit() 方法的源码:

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 的值小于 0 时,直接给 g.sem 赋值为 nil,表示不限制并发数量。

在调用 SetLimit() 方法时,g.sem 必须是一个空通道,否则程序会 panic

除去 SetLimit() 方法的判断逻辑代码,实际上 SetLimit() 方法就是创建一个大小为 n 的有缓冲 channel

SetLimit()TryGo() 通常一起使用。

04

总结

本文我们介绍 Go 方法提供的 errgroup 库,该库最近新增了控制并发数量的功能。

我们先介绍了三种使用方式,然后通过阅读源码,分析其实现原理。

errgroup 库的代码不多,建议感兴趣的读者朋友们阅读完整源码。

推荐阅读:

参考资料:

  1. https://github.com/golang/sync/blob/master/errgroup/errgroup.go
  2. https://pkg.go.dev/golang.org/x/sync@v0.1.0/errgroup
目录
相关文章
|
14天前
|
运维 监控 算法
监控局域网其他电脑:Go 语言迪杰斯特拉算法的高效应用
在信息化时代,监控局域网成为网络管理与安全防护的关键需求。本文探讨了迪杰斯特拉(Dijkstra)算法在监控局域网中的应用,通过计算最短路径优化数据传输和故障检测。文中提供了使用Go语言实现的代码例程,展示了如何高效地进行网络监控,确保局域网的稳定运行和数据安全。迪杰斯特拉算法能减少传输延迟和带宽消耗,及时发现并处理网络故障,适用于复杂网络环境下的管理和维护。
|
15天前
|
编译器 Go
揭秘 Go 语言中空结构体的强大用法
Go 语言中的空结构体 `struct{}` 不包含任何字段,不占用内存空间。它在实际编程中有多种典型用法:1) 结合 map 实现集合(set)类型;2) 与 channel 搭配用于信号通知;3) 申请超大容量的 Slice 和 Array 以节省内存;4) 作为接口实现时明确表示不关注值。此外,需要注意的是,空结构体作为字段时可能会因内存对齐原因占用额外空间。建议将空结构体放在外层结构体的第一个字段以优化内存使用。
|
19天前
|
存储 Go
Go 语言入门指南:切片
Golang中的切片(Slice)是基于数组的动态序列,支持变长操作。它由指针、长度和容量三部分组成,底层引用一个连续的数组片段。切片提供灵活的增减元素功能,语法形式为`[]T`,其中T为元素类型。相比固定长度的数组,切片更常用,允许动态调整大小,并且多个切片可以共享同一底层数组。通过内置的`make`函数可创建指定长度和容量的切片。需要注意的是,切片不能直接比较,只能与`nil`比较,且空切片的长度为0。
Go 语言入门指南:切片
|
18天前
|
开发框架 前端开发 Go
eino — 基于go语言的大模型应用开发框架(二)
本文介绍了如何使用Eino框架实现一个基本的LLM(大语言模型)应用。Eino中的`ChatModel`接口提供了与不同大模型服务(如OpenAI、Ollama等)交互的统一方式,支持生成完整响应、流式响应和绑定工具等功能。`Generate`方法用于生成完整的模型响应,`Stream`方法以流式方式返回结果,`BindTools`方法为模型绑定工具。此外,还介绍了通过`Option`模式配置模型参数及模板功能,支持基于前端和用户自定义的角色及Prompt。目前主要聚焦于`ChatModel`的`Generate`方法,后续将继续深入学习。
156 7
|
15天前
|
存储 缓存 监控
企业监控软件中 Go 语言哈希表算法的应用研究与分析
在数字化时代,企业监控软件对企业的稳定运营至关重要。哈希表(散列表)作为高效的数据结构,广泛应用于企业监控中,如设备状态管理、数据分类和缓存机制。Go 语言中的 map 实现了哈希表,能快速处理海量监控数据,确保实时准确反映设备状态,提升系统性能,助力企业实现智能化管理。
28 3
|
15天前
|
存储 缓存 安全
Go 语言中的 Sync.Map 详解:并发安全的 Map 实现
`sync.Map` 是 Go 语言中用于并发安全操作的 Map 实现,适用于读多写少的场景。它通过两个底层 Map(`read` 和 `dirty`)实现读写分离,提供高效的读性能。主要方法包括 `Store`、`Load`、`Delete` 等。在大量写入时性能可能下降,需谨慎选择使用场景。
|
19天前
|
存储 开发框架 Devops
eino — 基于go语言的大模型应用开发框架(一)
Eino 是一个受开源社区优秀LLM应用开发框架(如LangChain和LlamaIndex)启发的Go语言框架,强调简洁性、可扩展性和可靠性。它提供了易于复用的组件、强大的编排框架、简洁明了的API、最佳实践集合及实用的DevOps工具,支持快速构建和部署LLM应用。Eino不仅兼容多种模型库(如OpenAI、Ollama、Ark),还提供详细的官方文档和活跃的社区支持,便于开发者上手使用。
121 8
|
19天前
|
存储 算法 Go
Go语言实战:错误处理和panic_recover之自定义错误类型
本文深入探讨了Go语言中的错误处理和panic/recover机制,涵盖错误处理的基本概念、自定义错误类型的定义、panic和recover的工作原理及应用场景。通过具体代码示例介绍了如何定义自定义错误类型、检查和处理错误值,并使用panic和recover处理运行时错误。文章还讨论了错误处理在实际开发中的应用,如网络编程、文件操作和并发编程,并推荐了一些学习资源。最后展望了未来Go语言在错误处理方面的优化方向。
|
16天前
|
SQL 安全 Java
阿里双十一背后的Go语言实践:百万QPS网关的设计与实现
解析阿里核心网关如何利用Go协程池、RingBuffer、零拷贝技术支撑亿级流量。 重点分享: ① 如何用gRPC拦截器实现熔断限流; ② Sync.Map在高并发读写中的取舍。
|
17天前
|
存储 算法 安全
基于 Go 语言的公司内网管理软件哈希表算法深度解析与研究
在数字化办公中,公司内网管理软件通过哈希表算法保障信息安全与高效管理。哈希表基于键值对存储和查找,如用户登录验证、设备信息管理和文件权限控制等场景,Go语言实现的哈希表能快速验证用户信息,提升管理效率,确保网络稳定运行。
26 0