深入浅出golang errgroup.Group

谢权 · · 13217 次点击 · · 开始浏览

这是一个创建于的文章,其中的信息可能已经有所发展或是发生改变。

原生支持并发是Go语言最强大的特性,比如channels和goroutines。但是对于Go语言初学者尝试并接受并发概念并不是很轻松。
Go团队发布的第一个 goroutines 的管理工具是 sync.WaitGroup,WaitGroup会阻塞直到指定数量的goroutine已经完成执行,这是文档中的一个例子:

var wg sync.WaitGroup
var urls = []string{
 "http://www.golang.org/",
 "http://www.google.com/",
 "http://www.somestupidname.com/",
}
for _, url := range urls {
 // Increment the WaitGroup counter.
 wg.Add(1)
 // Launch a goroutine to fetch the URL.
 go func(url string) {
 // Decrement the counter when the goroutine completes.
 defer wg.Done()
 // Fetch the URL.
 http.Get(url)
 }(url)
}
// Wait for all HTTP fetches to complete.
wg.Wait()

varwg sync.WaitGroup

varurls=[]string{

"http://www.golang.org/",

"http://www.google.com/",

"http://www.somestupidname.com/",

}

for_,url:=rangeurls{

// Increment the WaitGroup counter.

wg.Add(1)

// Launch a goroutine to fetch the URL.

go func(url string){

// Decrement the counter when the goroutine completes.

defer wg.Done()

// Fetch the URL.

http.Get(url)

}(url)

}

// Wait for all HTTP fetches to complete.

wg.Wait()

WaitGroup 使你在处理并发任务时对goroutines的创建和停止的数量控制变得简单。每次你创建一个goroutine的时候只要调用Add()方法就可以了。当这个任务结束调用wg.Done()。等待所有的任务完成,调用 wg.Wait()。但是用 WatiGroup唯一的问题就是当你的goroutines出错时,你并不能知道出错的原因。

Extending sync.WaitGroup’s functionality

最近,Go团队在实验仓库中添加了一个名为sync.ErrGroup的新软件包。 sync.ErrGroup再sync.WaitGroup功能的基础上,增加了错误传递,以及在发生不可恢复的错误时取消整个goroutine集合,或者等待超时。我们同样来看一个示例:

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

vargerrgroup.Group

varurls=[]string{

"http://www.golang.org/",

"http://www.google.com/",

"http://www.somestupidname.com/",

}

for_,url:=rangeurls{

// 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)

iferr==nil{

resp.Body.Close()

}

returnerr

})

}

// Wait for all HTTP fetches to complete.

iferr:=g.Wait();err==nil{

fmt.Println("Successfully fetched all URLs.")

}

g.Go()方法不仅允许你传一个匿名的函数,而且还能捕获错误信息,你只要像这样返回一个错误 return err,这使开发者使用goroutines时开发效率显著提高。

为了测试sync.ErrGroup的所有功能,我写了一个小程序,用一个指定的模式递归搜索目录中的Go文件。这有助于在Go源代码树中查找已使用已弃用或更新的包的实例。要测试sync.ErrGroup的所有功能,我还为应用程序添加了超时设置在功能。如果达到时间限制,所有搜索和处理goroutine将被取消,程序将结束

当程序运行时,它会生成以下结果:

$ gogrep -timeout 1000ms . fmt 
gogrep.go
1 hits

$gogrep-timeout1000ms.fmt

gogrep.go

1hits

如果你使用的参数不正确的话,你会看到下面的输出

gogrep by Brian Ketelsen
Flags:
-timeout duration
 timeout in milliseconds (default 500ms)
Usage:
gogrep [flags] path pattern

gogrep by Brian Ketelsen

Flags:

-timeout duration

timeout inmilliseconds(default500ms)

Usage:

gogrep[flags]path pattern

How sync.ErrGroup makes application building easier

让我们看看我们是如何利用 sync.ErrGroup 来使程序写的更简单。我们将从一个 main() 函数开始。

package main
import (
 "bytes"
 "flag"
 "fmt"
 "io/ioutil"
 "log"
 "os"
 "path/filepath"
 "strings"
 "time"
 "golang.org/x/net/context"
 "golang.org/x/sync/errgroup"
)
func main() {
 duration := flag.Duration("timeout", 500*time.Millisecond, "timeout in milliseconds")
 flag.Usage = func() {
 fmt.Printf("%s by Brian Ketelsen\n", os.Args[0])
 fmt.Println("Usage:")
 fmt.Printf(" gogrep [flags] path pattern \n")
 fmt.Println("Flags:")
 flag.PrintDefaults()
 }
 flag.Parse()
 if flag.NArg() != 2 {
 flag.Usage()
 os.Exit(-1)
 }
 path := flag.Arg(0)
 pattern := flag.Arg(1)
 ctx, _ := context.WithTimeout(context.Background(), *duration)
 m, err := search(ctx, path, pattern)
 if err != nil {
 log.Fatal(err)
 }
 for _, name := range m {
 fmt.Println(name)
 }
 fmt.Println(len(m), "hits")
}
func search(ctx context.Context, root string, pattern string) ([]string, error) {
 g, ctx := errgroup.WithContext(ctx)
 paths := make(chan string, 100)
 // get all the paths
 g.Go(func() error {
 defer close(paths)
 return filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
 if err != nil {
 return err
 }
 if !info.Mode().IsRegular() {
 return nil
 }
 if !info.IsDir() &amp;&amp; !strings.HasSuffix(info.Name(), ".go") {
 return nil
 }
 select {
 case paths &lt;- path:
 case &lt;-ctx.Done():
 return ctx.Err()
 }
 return nil
 })
 })
 c := make(chan string, 100)
 for path := range paths {
 p := path
 g.Go(func() error {
 data, err := ioutil.ReadFile(p)
 if err != nil {
 return err
 }
 if !bytes.Contains(data, []byte(pattern)) {
 return nil
 }
 select {
 case c &lt;- p:
 case &lt;-ctx.Done():
 return ctx.Err()
 }
 return nil
 })
 }
 go func() {
 g.Wait()
 close(c)
 }()
 var m []string
 for r := range c {
 m = append(m, r)
 }
 return m, g.Wait()
}

100

101

packagemain

import(

"bytes"

"flag"

"fmt"

"io/ioutil"

"log"

"os"

"path/filepath"

"strings"

"time"

"golang.org/x/net/context"

"golang.org/x/sync/errgroup"

)

func main(){

duration:=flag.Duration("timeout",500*time.Millisecond,"timeout in milliseconds")

flag.Usage=func(){

fmt.Printf("%s by Brian Ketelsen\n",os.Args[0])

fmt.Println("Usage:")

fmt.Printf(" gogrep [flags] path pattern \n")

fmt.Println("Flags:")

flag.PrintDefaults()

}

flag.Parse()

ifflag.NArg()!=2{

flag.Usage()

os.Exit(-1)

}

path:=flag.Arg(0)

pattern:=flag.Arg(1)

ctx,_:=context.WithTimeout(context.Background(),*duration)

m,err:=search(ctx,path,pattern)

iferr!=nil{

log.Fatal(err)

}

for_,name:=rangem{

fmt.Println(name)

}

fmt.Println(len(m),"hits")

}

func search(ctx context.Context,root string,pattern string)([]string,error){

g,ctx:=errgroup.WithContext(ctx)

paths:=make(chan string,100)

// get all the paths

g.Go(func()error{

defer close(paths)

returnfilepath.Walk(root,func(path string,info os.FileInfo,err error)error{

iferr!=nil{

returnerr

}

if!info.Mode().IsRegular(){

returnnil

}

if!info.IsDir()&&!strings.HasSuffix(info.Name(),".go"){

returnnil

}

select{

casepaths<-path:

case<-ctx.Done():

returnctx.Err()

}

returnnil

})

c:=make(chan string,100)

forpath:=rangepaths{

p:=path

g.Go(func()error{

data,err:=ioutil.ReadFile(p)

iferr!=nil{

returnerr

}

if!bytes.Contains(data,[]byte(pattern)){

returnnil

}

select{

casec<-p:

case<-ctx.Done():

returnctx.Err()

}

returnnil

})

}

go func(){

g.Wait()

close(c)

}()

varm[]string

forr:=rangec{

m=append(m,r)

}

returnm,g.Wait()

}

在这行代码以前,是对命令行传进来的参数做了解析。这行以后才是我们真正感兴趣的代码。

ctx, _ := context.WithTimeout(context.Background(), *duration)

ctx,_:=context.WithTimeout(context.Background(),*duration)

我创建了一个context.Context并且为它添加了超时设置,当超时时间到了,"ctx"将接收到channel的超时警告。WithTimeout同样也会返回一个取消的方法,但是我们不需要,所以用 "_" 来忽略掉了。
下面的search() 方法的有context,search path,和 search pattern。最后把找到的文件和数量输出。

参考文献:https://www.oreilly.com/learning/run-strikingly-fast-parallel-file-searches-in-go-with-sync-errgroup
https://godoc.org/golang.org/x/sync/errgroup#example-Group–Parallel

深入浅出golang errgroup.Group

有疑问加站长微信联系(非本文作者)

本文来自:谢权SELF

感谢作者:谢权

查看原文:深入浅出golang errgroup.Group

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

关注微信

13217 次点击

上一篇:Eclipse源码阅读:Go To Resource

下一篇:OSX平台下跨平台编译Go程序(交叉编译)

http context goroutine https

0 回复

暂无回复

添加一条新回复 (您需要后才能回复没有账号 ?)

请尽量让自己的回复能够对别人有帮助
支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
图片支持拖拽、截图粘贴等方式上传

用户登录

Go今日面试题

(追記) (追記ここまで)

今日阅读排行

加载中

(追記) (追記ここまで)

一周阅读排行

加载中

关注我

扫码关注领全套学习资料关注微信公众号
加入 QQ 群:
- 192706294(已满)
- 731990104(已满)
- 798786647(已满)
- 729884609(已满)
- 977810755(已满)
- 815126783(已满)
- 812540095(已满)
- 1006366459(已满)
- 692541889
关注微信公众号
加入微信群:liuxiaoyan-s,备注入群
也欢迎加入知识星球 Go粉丝们(免费)

给该专栏投稿写篇新文章

每篇文章有总共有 5 次投稿机会

深入浅出golang errgroup.Group

Extending sync.WaitGroup’s functionality

How sync.ErrGroup makes application building easier

用户登录

今日阅读排行

一周阅读排行

关注我

给该专栏投稿 写篇新文章

收入到我管理的专栏 新建专栏

Extending sync.WaitGroup’s functionality

How sync.ErrGroup makes application building easier

给该专栏投稿写篇新文章

收入到我管理的专栏新建专栏