Go并发控制errgroup.Group的实现示例

errgroup：不止于同步，更优雅地处理并发错误

在Go语言的并发世界里，errgroup 是一个绕不开的利器。它来自 golang.org/x/sync/errgroup，核心使命是为一组执行共同子任务的协程（goroutines）提供同步、错误传播和上下文取消的能力。简单说，它让管理一群“工人”并处理他们可能带来的“麻烦”变得井井有条。

从 WaitGroup 到 errGroup：解决了什么痛点？

但凡需要等待多个并发任务全部完成后再继续的场景，开发者们第一时间想到的往往是 sync.WaitGroup。它确实是个可靠的同步原语，但有个明显的短板：它只管“等”，不管“错”。任何一个goroutine内部发生的错误，都无法通过WaitGroup本身反馈给主调方。

这时，errgroup.Group 的价值就凸显出来了。你可以把它看作是WaitGroup的“增强版”，它在提供同步等待能力的基础上，额外增加了两大法宝：对返回错误任务的处理能力，以及限制协程并发数的能力。其使用方法与WaitGroup一脉相承，但内部封装了Add和Wait，巧妙地解决了错误无法返回的难题。

来看一个基础示例，感受一下它是如何工作的：

package main
import (
 "fmt"
 "time"
 "golang.org/x/sync/errgroup"
)
func main() {
 g := &errgroup.Group{}
 for i := 0; i < 5; i++ {
  index := i
  g.Go(func() error {
   fmt.Printf("start to execute the %d gorouting\n", index)
   time.Sleep(time.Duration(index) * time.Second)
   if index%2 == 0 {
    return fmt.Errorf("something has failed on grouting:%d", index)
   }
   fmt.Printf("gorouting:%d end\n", index)
   return nil
  })
 }
 if err := g.Wait(); err != nil {
  fmt.Println(err)
 }
}
// Output:
// start to execute the 4 gorouting
// start to execute the 1 gorouting
// start to execute the 0 gorouting
// start to execute the 2 gorouting
// start to execute the 3 gorouting
// gorouting:1 end
// gorouting:3 end
// something has failed on grouting:0

运行这段代码，你会发现一个关键特性：当多个goroutine出错时，errgroup只会返回第一个出错的goroutine的错误信息。这符合快速失败（fail-fast）的常见设计原则。另一个需要留意的点是，无论是否有协程执行失败，Wait()方法都会忠实地等待所有协程执行完毕，确保资源被妥善清理。

进阶特性：上下文与并发控制

除了基础功能，errgroup还提供了更贴合现代Go开发范式的特性。首先是原生支持context，方便进行跨协程的取消信号传播和超时控制：

g, _ := errgroup.WithContext(context.Background()) // 支持 context

其次，Wait()方法的设计考虑到了复用性，它可以被多次调用，并且每次都能得到相同的错误信息，这为一些需要重复检查结果的场景提供了便利：

 ...
 if err := g.Wait(); err != nil {
  fmt.Println(err)
 }
 if err := g.Wait(); err != nil { // 可再次调用 Wait，依然可以得到 group 的 error 信息
  fmt.Println(err)
 }

核心利器：限制最大并发数

在实际生产环境中，无限制地创建goroutine可能导致资源耗尽。errgroup的SetLimit()方法正是为此而生。它用于限制该组中同时处于活动状态（即正在处理业务）的goroutine的最大数量。

通过一个简单的任务处理示例，可以清晰地看到它的效果：

package main
import (
 "log"
 "time"
 "golang.org/x/sync/errgroup"
)
func main() {
 jobs := make(chan int, 10)
 go func() {
  for i := 0; i < 8; i++ {
   jobs <- i + 1
  }
  close(jobs)
 }()
 eg:= &errgroup.Group{}
 eg.SetLimit(3)
 for j := range jobs {
  j := j
  eg.Go(func() error {
   log.Printf("handle job: %d\n", j)
   time.Sleep(2 * time.Second)
   return nil
  })
 }
 eg.Wait()
}

在这个例子中，我们创建了8个任务，但通过SetLimit(3)，确保了同一时间最多只有3个goroutine在忙碌地处理任务。

Output:

2024/12/17 18:28:19 handle job: 3

2024/12/17 18:28:19 handle job: 1

2024/12/17 18:28:19 handle job: 2

2024/12/17 18:28:21 handle job: 4

2024/12/17 18:28:21 handle job: 6

2024/12/17 18:28:21 handle job: 5

2024/12/17 18:28:23 handle job: 7

2024/12/17 18:28:23 handle job: 8

仔细观察输出结果的时间戳，规律一目了然：任务1、2、3同时开始，2秒后任务4、5、6接棒，再2秒后任务7、8最后完成。这完美印证了并发数被限制在3个。其内部实现原理并不复杂，本质上是使用了一个带缓冲的channel作为计数信号量（counting semaphore）来控制并发许可的发放，这种实现既高效又简洁。

话说回来，将错误处理、并发控制和同步等待三者优雅地结合，正是errgroup在Go并发工具箱中占据一席之地的原因。对于需要精细化管理并发任务流的场景，它无疑是一个值得深入掌握的标准组件。