一、概述

golang官方版本：golang.org/x/sync/errgroup
改进版：https://github.com/chitian-victor/errgroup

核心方法：

golang官方版本的介绍

errgroup.WithContext，初始化Group变量，生成新的带有cancel方法的context变量

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

Group.Go，发起协程，监听到error产生，调用cancel方法

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

Group.Wait，等待所有协程执行完毕

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

用法示例：

func main() {
    ctx := context.Background()
    // 1. init group
    group, newCtx := errgroup.WithContext(ctx)

    // 2. initiate goroutine
    group.Go(func() error {
       // do something 1
    })
    group.Go(func() error {
       // do something 2
    })

    // 3. wait for all goroutine to return
    if err := group.Wait(); err != nil {
        // handle error
    }
}

对比

相较于golang官方版本

优点：

支持捕获panic并报错返回
解除对context类cancel方法的依赖

缺点：

协程加倍，所以一般适用于协程数量不高的场景「go默认协程最多1万个」

为什么“解除对context类cancel方法的依赖”是优点，如何体现？
errgroup有一个「快速响应」的特性：发起10个协程进行rpc调用，那么其中1个协程的rpc调用失败报错返回，另外9个中未执行完毕的协程也会立刻返回，达成「快速响应」效果

想要让自己的程序具备「快速响应」的特性，前提要实现对context cancel事件的监听，而一些 golang 的 rpc框架「比如kitex、gorm」刚好有实现了这套监听的逻辑，所以才能触发「快速响应」；其次，如果我们只是将自定义的计算逻辑放入其中，同样也要实现对context cancel事件的监听才能达到「快速响应」的效果

所以，改进版直接解除调用对context cancel事件的依赖，维护一个新变量，同时主动为用户监听该变量的‘cancel’事件

实现

// Package chitian/errgroup is an extension of the sync/errgroup which has this header comment:
//
// 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

import (
    "context"
    "log"
    "runtime"
    "sync"
)
type Error string

func (e Error) Error() string {
    return string(e)
}

const (
    PanicError = Error("errgroup: goroutine panic")
)
// 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 {
    doneOnce sync.Once
    done     chan struct{}

    wg      sync.WaitGroup
    errOnce sync.Once
    err     error

    ch chan struct{}
}

// default cancel on error.
func WithCancel() *Group {
    return &Group{done: make(chan struct{})}
}

// Not Recommended, just to be compatibility with package "golang.org/x/sync/errgroup"
func WithContext(ctx context.Context) (*Group, context.Context) {
    return &Group{done: make(chan struct{})}, ctx
}

// set max goroutine to work.
func (g *Group) SetLimit(n int) {
    g.ch = make(chan struct{}, n)
}

// close channel only once
func (g *Group) cancel() {
    if g.done == nil {
        return
    }
    g.doneOnce.Do(func() {
        close(g.done)
    })
}

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

// 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()
    g.cancel()
    if g.ch != nil {
        close(g.ch)
    }
    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) {
    if g.ch != nil {
        select {
        case <-g.done:
            return
        case g.ch <- struct{}{}:
            break
        }
    }

    g.wg.Add(1)
    go func() {
        defer g.done_()
        localDone := make(chan error, 2)
        go func() {
            defer func() {
                if err := recover(); err != nil {
                    const size = 64 << 10
                    buf := make([]byte, size)
                    buf = buf[:runtime.Stack(buf, false)]
                    log.Printf("panic: err=%s\n%s", err, string(buf))
                    localDone <- PanicError
                }
                close(localDone)
            }()
            if err := f(); err != nil {
                localDone <- err
            }
        }()

        select {
        case err := <-localDone:
            if err != nil {
                g.errOnce.Do(func() {
                    g.err = err
                })
                g.cancel()
            }
        case <-g.done: // when g.done is nil, this case will be blocked
        }
    }()
}

errgroup 改进版

一、概述

核心方法：

对比

实现