Taskgroup

A simple and useful goroutine concurrent library.

Installation

go get github.com/anthhub/taskgroup

Usage

count:=100// create a taskgroup// set max error count to 1g:=New(&Option{MaxErrorCount:1})
fori:=0; i<count; i++ {
g.Go(func() (interface{}, error) {
// your work functionreturnworker()
})
}
// g.Fed() declare the end of tasks producing// // g.Result() will receive a message when a task of the group return an error or// till all tasks finish.ifp:=<-g.Fed().Result(); p.Err!=nil {
return
}
// ...

Advanced Usage

Useful Options

count:=100// configure group timeoutctx, _:=context.WithTimeout(context.Background(), time.Second)
// create a taskgroup with contextg, ctx=WithContext(ctx, &Option{
// limit the count of goroutine workers; default is infinityLimit: 5,
// set max error count to 5; default is infinityMaxErrorCount: 5,
// disable recover panic; default will recover panicDisableRecover: true,
})
// the loop need be wrapped by a goroutine when the limit less than your tasks count,// else dead lock will be created.gofunc() {
fori:=0; i<count; i++ {
g.Go(func() (interface{}, error) {
// your work function that return data and error// panic will be recover and just return a errorreturnworker(ctx)
})
}
// it is to tell consumer that consuming the all tasks and then close itself.// // it is very important else the consuming never end without g.Fed() when all // tasks have finished.g.Fed()
}()
// you can directly for-range g.Result(), it will break the loop when all tasks id finished// or error count is to 5.forp:=rangeg.Result() {
ifp.Err!=nil {
glog.Errorf("error")
}
// you can cancel the group and break the loop in advanced when you want.if [condition] {
g.Cancel()
break
}
data:= p.Data// consume data from workerconsume(p.Data)
// ...
}
// ...

Producer & Consumer Mode

funcmain() {
g:=provider()
consumer(g)
// some logic ...delay(10)
// cancel provider and consumerg.Cancel()
}
funcprovider() Group {
g:=New(&Option{MaxErrorCount: 1})
gofunc() {
for {
select {
case<-g.Ctx().Done():
returndefault:
g.Go(func() (interface{}, error) {
// get group inner ctxreturnworker(g.Ctx())
})
}
delay(1)
}
}()
returng
}
funcconsumer(gGroup) { gofunc() { forp:=rangeg.Result() { // it is just consuming all message from provider till provider want to stop, so	// g.Fed() is not necessary	ifp.Err!=nil {
continue
}
// consume data from providerconsume(p.Data)
} }()
}

If you want to learn more about taskgroup, you can read test cases and source code.