详解Go语言中ErrGroup的使用

func WithContext(ctx context.Context) (*Group, context.Context)
func (g *Group) Go(f func() error)
func (g *Group) Wait() error

package main
import (
	"fmt"
	"golang.org/x/sync/errgroup"
	"net/http"
)
func main() {
	var urls = []string{
		"http://www.golang.org/",
		"http://www.baidu.com/",
		"http://www.noexist11111111.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!")
	}
}

package main
import (
	"fmt"
	"golang.org/x/sync/errgroup"
	"log"
	"time"
)
func main() {
	var eg errgroup.Group
	for i := 0; i < 100; i++ {
		i := i
		eg.Go(func() error {
			time.Sleep(2 * time.Second)
			if i > 90 {
				fmt.Println("Error:", i)
				return fmt.Errorf("Error occurred: %d", i)
			}
			fmt.Println("End:", i)
			return nil
		})
	}
	if err := eg.Wait(); err != nil {
		log.Fatal(err)
	}
}

package main
import (
	"context"
	"fmt"
	"golang.org/x/sync/errgroup"
	"log"
	"time"
)
func main() {
	eg, ctx := errgroup.WithContext(context.Background())
	for i := 0; i < 100; i++ {
		i := i
		eg.Go(func() error {
			time.Sleep(2 * time.Second)
			select {
			case <-ctx.Done():
				fmt.Println("Canceled:", i)
				return nil
			default:
				if i > 90 {
					fmt.Println("Error:", i)
					return fmt.Errorf("Error: %d", i)
				}
				fmt.Println("End:", i)
				return nil
			}
		})
	}
	if err := eg.Wait(); err != nil {
		log.Fatal(err)
	}
}

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 func()
    wg sync.WaitGroup
    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
}
// 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 {
                    g.cancel()
                }
            })
        }
    }()
}

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!")
}

package main
import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)
const (
	M = 2
	N = 8
)
func main() {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*50)
	defer cancel()
	result := make([]int, N+1)
	errCh := make(chan error, 1)
	var firstSendErr int32
	wg := new(sync.WaitGroup)
	done := make(chan struct{}, 1)
	limit := make(chan struct{}, M)
	for i := 1; i <= N; i++ {
		limit <- struct{}{}
		var quit bool
		select {
		// context已经被cancel,不需要起新的goroutine了
		case <-ctx.Done():
			quit = true
		default:
		}
		if quit {
			break
		}
		wg.Add(1)
		go func(x int) {
			defer func() {
				wg.Done()
				<-limit
			}()
			if ret, err := doTask(ctx, x); err != nil {
				if atomic.CompareAndSwapInt32(&firstSendErr, 0, 1) {
					errCh <- err
					// cancel其他的请求
					cancel()
				}
			} else {
				result[x] = ret
			}
		}(i)
	}
	go func() {
		wg.Wait()
		close(done)
	}()
	select {
	case err := <-errCh:
		handleErr(err, result[1:])
		<-done
	case <-done:
		if len(errCh) > 0 {
			err := <-errCh
			handleErr(err, result[1:])
			return
		}
		fmt.Println("success handle all task:", result[1:])
	}
}
func handleErr(err error, result []int) {
	fmt.Println("task err occurs: ", err, "result", result)
}
func doTask(ctx context.Context, i int) (ret int, err error) {
	fmt.Println("task start", i)
	defer func() {
		fmt.Println("task done", i, "err", err)
	}()
	select {
	// 模拟处理任务时间
	case <-time.After(time.Second * time.Duration(i)):
	// 处理任务要支持被context cancel,不然就一直等到处理完再返回了
	case <-ctx.Done():
		fmt.Println("task canceled", i)
		return -1, ctx.Err()
	}
	// 模拟出现错误
	if i == 6 {
		return -1, errors.New("err test")
	}
	return i, nil
}