golang goroutine 无法抛错就用 errgroup

一般在golang 中想要并发运行业务时会直接开goroutine，关键字go ,但是直接go的话函数是无法对返回数据进行处理error的。

解决方案：

• 初级版本：

一般是直接在出错的地方打入log日志,将出的错误记录到日志文件中，也可以集合日志收集系统直接将该错误用邮箱或者办公软件发送给你如：钉钉机器人+graylog.

• 中级版本

当然你也可以自己在log包里封装好可以接受channel。

利用channel通道，将go中出现的error传入到封装好的带有channel接受器的log包中，进行错误收集或者通知通道接受return出来即可

• 终极版本 errgroup

这个包是google对go的一个扩展包：

golang.org/x/sync/errgroup

怎么调用

我们直接看看官方test的demo调用：

func ExampleGroup_justErrors() {
    var g errgroup.Group
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/",
    }
    for _, url := range urls {
        // Launch a goroutine to fetch the URL.
        url := url // https://golang.org/doc/faq#closures_and_goroutines
        g.Go(func() error {
            // Fetch the URL.
            resp, err := http.Get(url)
            if err == nil {
                resp.Body.Close()
            }
            return err
        })
    }
    // Wait for all HTTP fetches to complete.
    if err := g.Wait(); err == nil {
        fmt.Println("Successfully fetched all URLs.")
    }
}

很简单的一个并发爬虫网页请求，请求中只要有一个有错误就会返回error。

再来看一种代有上下文context的调用:

func TestWithContext(t *testing.T) {
    errDoom := errors.New("group_test: doomed")

    cases := []struct {
        errs []error
        want error
    }{
        {want: nil},
        {errs: []error{nil}, want: nil},
        {errs: []error{errDoom}, want: errDoom},
        {errs: []error{nil, errDoom}, want: errDoom},
    }

    for _, tc := range cases {
        g, ctx := errgroup.WithContext(context.Background())

        for _, err := range tc.errs {
            err := err
            g.Go(func() error {
                log.Error(err) // 当此时的err = nil 时，g.Go不会将 为nil 的 err 放入g.err中
                return err
            })
        }
        err := g.Wait() // 这里等待所有Go跑完即add==0时，此处的err是g.err的信息。
        log.Error(err)
        log.Error(tc.want)
        if err != tc.want {
            t.Errorf("after %T.Go(func() error { return err }) for err in %v\n"+
                "g.Wait() = %v; want %v",
                g, tc.errs, err, tc.want)
        }

        canceled := false
        select {
        case <-ctx.Done():
            // 由于上文中内部调用了cancel(),所以会有Done()接受到了消息
            // returns an error or ctx.Done is closed 
            // 在当前工作完成或者上下文被取消之后关闭
            canceled = true
        default:
        }
        if !canceled {
            t.Errorf("after %T.Go(func() error { return err }) for err in %v\n"+
                "ctx.Done() was not closed",
                g, tc.errs)
        }
    }
}

关于上下文的知识补充可以看链接：

上下文 Context

这里 调用了 errgroup.WithContext， 用于控制每一个goroutined的生理周期。
在不同的 Goroutine 之间同步请求特定的数据、取消信号以及处理请求的截止日期

总的来说api的使用还是很简单的，再来看看源码包吧

源码解析

// Copyright 2016 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

// Package errgroup provides synchronization, error propagation, and Context
// cancelation for groups of goroutines working on subtasks of a common task.
package errgroup

import (
    "context"
    "sync"
)

// A Group is a collection of goroutines working on subtasks that are part of
// the same overall task.
//
// A zero Group is valid and does not cancel on error.
type Group struct {
    // cancel 初始化时会直接扔一个context.WithCancel(ctx)的cancel进入，然后通过它来进行removeChild removes a context from its parent
    // 这样真正调用其实是：func() { c.cancel(true, Canceled) }
    cancel func()
    
    // 包含了个 WaitGroup用于同步等待所有Gorourine执行
    
    wg sync.WaitGroup 

    // go语言中特有的单例模式，利用原子操作进行锁定值判断
    
    errOnce sync.Once
    
    err     error
}

// WithContext returns a new Group and an associated Context derived from ctx.
//
// The derived Context is canceled the first time a function passed to Go
// returns a non-nil error or the first time Wait returns, whichever occurs
// first.

func WithContext(ctx context.Context) (*Group, context.Context) {
    ctx, cancel := context.WithCancel(ctx)
    return &Group{cancel: cancel}, ctx
}

// Wait blocks until all function calls from the Go method have returned, then
// returns the first non-nil error (if any) from them.

func (g *Group) Wait() error {
    g.wg.Wait()
    if g.cancel != nil {
        g.cancel()
    }
    return g.err 
    // 这里返回的error其实是从众多goroutine中返回第一个非nil的错误信息，所以这个错误信息如果全部都是一样的话，你是不知道到底是哪个goroutine中报的错，应该在goroutine内部就写清楚错误信息的别分类似可以加入id值这种。
}

// Go calls the given function in a new goroutine.
//
// The first call to return a non-nil error cancels the group; its error will be
// returned by Wait.
func (g *Group) Go(f func() error) {
    g.wg.Add(1)

    go func() {
        defer g.wg.Done()

        if err := f(); err != nil {
            g.errOnce.Do(func() {
                g.err = err
                if g.cancel != nil {
                    // 如果出现第一个err为非nil就会去调用关闭接口，即关闭后面的所有子gorourine
                    g.cancel()
                }
            })
        }
    }()
}

开源引用demo

这里介绍一个 bilibili errgroup 包：

bilbil/kratos

另外 官方包golang.org/sync/errgroup 的 test 也可以看看。