Golang sync包中errgroup的使用详解

WaitGroup 主要用于控制任务组下的并发子任务。它的具体做法就是，子任务 goroutine 执行前通过 Add 方法添加任务数目，子任务 goroutine 结束时调用 Done 标记已完成任务数，主任务 goroutine 通过 Wait 方法等待所有的任务完成后才能执行后续逻辑

package main
  
 import (
     "net/http"
     "sync"
 )
  
 func main() {
    var wg sync.WaitGroup
    var urls = []string{
        "http://www.golang.org/",
        "http://www.baidu.com/",
        "http://www.bokeyuan12111.com/",
    }
    for _, url := range urls {
        wg.Add(1)
        go func(url string) {
            defer wg.Done()
            resp, err := http.Get(url)
            if err != nil {
                return
            }
            resp.Body.Close()
        }(url)
    }
    wg.Wait()
}

但在实际的项目代码中，子任务 goroutine 的执行并不总是顺风顺水，它们也许会产生 error。而 WaitGroup 并没有告诉我们在子 goroutine 发生错误时，如何将其抛给主任务 groutine。这个时候可以考虑使用 errgroup

package main
  
 import (
     "fmt"
     "net/http"
  
     "golang.org/x/sync/errgroup"
 )
  
func main() {
    var urls = []string{
        "http://www.golang.org/",
        "http://www.baidu.com/",
        "http://www.bokeyuan12111.com/",
    }
    g := new(errgroup.Group)
    for _, url := range urls {
        url := url
        g.Go(func() error {
            resp, err := http.Get(url)
            if err != nil {
                fmt.Println(err)
                return err
            }
            fmt.Printf("get [%s] success: [%d] \n", url, resp.StatusCode)
            return resp.Body.Close()
        })
    }
    if err := g.Wait(); err != nil {
        fmt.Println(err)
    } else {
        fmt.Println("All success!")
    }
}

get [http://www.baidu.com/] success: [200]
Get "http://www.bokeyuan12111.com/": dial tcp: lookup www.bokeyuan12111.com: no such host
Get "http://www.golang.org/": dial tcp 142.251.42.241:80: i/o timeout
Get "http://www.bokeyuan12111.com/": dial tcp: lookup www.bokeyuan12111.com: no such host

errgroup 上下文取消

package main
 
import (
    "context"
    "fmt"
 
    "golang.org/x/sync/errgroup"
)
 
func main() {
 
    g, ctx := errgroup.WithContext(context.Background())
    dataChan := make(chan int, 20)
 
    // 数据生产端任务子 goroutine
    g.Go(func() error {
        defer close(dataChan)
        for i := 1; ; i++ {
            if i == 10 {
                return fmt.Errorf("data 10 is wrong")
            }
            dataChan <- i
            fmt.Println(fmt.Sprintf("sending %d", i))
        }
    })
 
    // 数据消费端任务子 goroutine
    for i := 0; i < 3; i++ {
        g.Go(func() error {
            for j := 1; ; j++ {
                select {
                case <-ctx.Done():
                    return ctx.Err()
                case number := <-dataChan:
                    fmt.Println(fmt.Sprintf("receiving %d", number))
                }
            }
        })
    }
 
    // 主任务 goroutine 等待 pipeline 结束数据流
    err := g.Wait()
    if err != nil {
        fmt.Println(err)
    }
    fmt.Println("main goroutine done!")
}

sending 1
sending 2
sending 3
sending 4
sending 5
sending 6
sending 7
sending 8
sending 9
receiving 1
receiving 3
receiving 2
receiving 4
data 10 is wrong
main goroutine done!