kratos: github.com/go-kratos/kratos/pkg/sync/errgroup Index | Examples | Files

package errgroup

import "github.com/go-kratos/kratos/pkg/sync/errgroup"

Package errgroup provides synchronization, error propagation, and Context errgroup 包为一组子任务的 goroutine 提供了 goroutine 同步,错误取消功能.

errgroup 包含三种常用方式

1、直接使用 此时不会因为一个任务失败导致所有任务被 cancel:

g := &errgroup.Group{}
g.Go(func(ctx context.Context) {
	// NOTE: 此时 ctx 为 context.Background()
	// do something
})

2、WithContext 使用 WithContext 时不会因为一个任务失败导致所有任务被 cancel:

g := errgroup.WithContext(ctx)
g.Go(func(ctx context.Context) {
	// NOTE: 此时 ctx 为 errgroup.WithContext 传递的 ctx
	// do something
})

3、WithCancel 使用 WithCancel 时如果有一个人任务失败会导致所有*未进行或进行中*的任务被 cancel:

g := errgroup.WithCancel(ctx)
g.Go(func(ctx context.Context) {
	// NOTE: 此时 ctx 是从 errgroup.WithContext 传递的 ctx 派生出的 ctx
	// do something
})

设置最大并行数 GOMAXPROCS 对以上三种使用方式均起效 NOTE: 由于 errgroup 实现问题,设定 GOMAXPROCS 的 errgroup 需要立即调用 Wait() 例如:

g := errgroup.WithCancel(ctx)
g.GOMAXPROCS(2)
// task1
g.Go(func(ctx context.Context) {
	fmt.Println("task1")
})
// task2
g.Go(func(ctx context.Context) {
	fmt.Println("task2")
})
// task3
g.Go(func(ctx context.Context) {
	fmt.Println("task3")
})
// NOTE: 此时设置的 GOMAXPROCS 为2, 添加了三个任务 task1, task2, task3 此时 task3 是不会运行的!
// 只有调用了 Wait task3 才有运行的机会
g.Wait() // task3 运行

Index

Examples

Package Files

doc.go errgroup.go

type Group Uses

type Group struct {
    // contains filtered or unexported fields
}

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.

Code:

g := WithCancel(context.Background())
g.Go(func(ctx context.Context) error {
    return fakeRunTask(ctx)
})
g.Go(func(ctx context.Context) error {
    return fakeRunTask(ctx)
})
if err := g.Wait(); err != nil {
    // handle err
}

Code:

g := WithContext(context.Background())
g.Go(func(ctx context.Context) error {
    return fakeRunTask(ctx)
})
g.Go(func(ctx context.Context) error {
    return fakeRunTask(ctx)
})
if err := g.Wait(); err != nil {
    // handle err
}

Code:

g := Group{}
g.Go(func(context.Context) error {
    return fakeRunTask(context.Background())
})
g.Go(func(context.Context) error {
    return fakeRunTask(context.Background())
})
if err := g.Wait(); err != nil {
    // handle err
}

JustErrors illustrates the use of a Group in place of a sync.WaitGroup to simplify goroutine counting and error handling. This example is derived from the sync.WaitGroup example at https://golang.org/pkg/sync/#example_WaitGroup.

Code:

var g 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(context.Context) 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.")
}

Code:

g := Group{}
// set max concurrency
g.GOMAXPROCS(2)
g.Go(func(ctx context.Context) error {
    return fakeRunTask(context.Background())
})
g.Go(func(ctx context.Context) error {
    return fakeRunTask(context.Background())
})
if err := g.Wait(); err != nil {
    // handle err
}

Parallel illustrates the use of a Group for synchronizing a simple parallel task: the "Google Search 2.0" function from https://talks.golang.org/2012/concurrency.slide#46, augmented with a Context and error-handling.

Code:

Google := func(ctx context.Context, query string) ([]Result, error) {
    g := WithContext(ctx)

    searches := []Search{Web, Image, Video}
    results := make([]Result, len(searches))
    for i, search := range searches {
        i, search := i, search // https://golang.org/doc/faq#closures_and_goroutines
        g.Go(func(context.Context) error {
            result, err := search(ctx, query)
            if err == nil {
                results[i] = result
            }
            return err
        })
    }
    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

results, err := Google(context.Background(), "golang")
if err != nil {
    fmt.Fprintln(os.Stderr, err)
    return
}
for _, result := range results {
    fmt.Println(result)
}

Output:

web result for "golang"
image result for "golang"
video result for "golang"

func WithCancel Uses

func WithCancel(ctx context.Context) *Group

WithCancel create a new Group and an associated Context derived from ctx.

given function from Go will receive context derived from this 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 Uses

func WithContext(ctx context.Context) *Group

WithContext create a Group. given function from Go will receive this context,

func (*Group) GOMAXPROCS Uses

func (g *Group) GOMAXPROCS(n int)

GOMAXPROCS set max goroutine to work.

func (*Group) Go Uses

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

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 (*Group) Wait Uses

func (g *Group) Wait() error

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

Package errgroup imports 4 packages (graph) and is imported by 3 packages. Updated 2020-10-08. Refresh now. Tools for package owners.