README
¶
Parallel
A golang parallel library, used for business logic aggregation and refactory without changing function declaration.
Install
go get github.com/buptmiao/parallel
Usage
eg.1
There are three methods: testjobA, testjobB, testjobC, execute them in parallel:
import (
"github.com/buptmiao/parallel"
)
func testJobA() string {
return "job"
}
func testJobB(x, y int) int {
return x + y
}
func testJobC(x int) int {
return -x
}
func main() {
var s string
var x, y int
p := parallel.NewParallel()
p.Register(testJobA).SetReceivers(&s)
p.Register(testJobB, 1, 2).SetReceivers(&x)
p.Register(testJobC, 3).SetReceivers(&y)
// block here
p.Run()
if s != "job" || x != 3 || y != -3{
panic("unexpected result")
}
}
eg.2
Let's see a little complex case, there are three parallel jobs: jobA, jobB, jobC and a final Job which aggregates the result. The final depends on jobA and middle which depends on jobB and jobC.
jobA jobB jobC
\ \ /
\ \ /
\ middle
\ /
\ /
final
Refer to the demo below:
import (
"github.com/buptmiao/parallel"
)
type middle struct {
B int
C int
}
type testResult struct {
A string
M middle
}
func testJobA() string {
return "job"
}
func testJobB(x, y int) int {
return x + y
}
func testJobC(x int) int {
return -x
}
func testFinal(s *string, m *middle) testResult {
return testResult{
*s, *m,
}
}
func main() {
var m middle
var s string
var res testResult
p := parallel.NewParallel()
// Create a child 1
child1 := p.NewChild()
child1.Register(testJobA).SetReceivers(&s)
// Create another child 2
child2 := p.NewChild()
child2.Register(testJobB, 1, 2).SetReceivers(&m.B)
child2.Register(testJobC, 2).SetReceivers(&m.C)
p.Register(testFinal, &s, &m).SetReceivers(&res)
// block here
p.Run()
expect := testResult{
"job",
middle{
3, -2,
},
}
if res != expect {
panic("unexpected result")
}
}
eg.3
By default, Parallel will ignore panics of jobs. But parallel supports customized exception handler, which is used for dealing with unexpected panics. For example, alerting or logging.
// handle the panic
func exceptionHandler(topic string, e interface{}) {
fmt.Println(topic, e)
}
// will panic
func exceptionJob() {
var a map[string]int
//assignment to entry in nil map
a["123"] = 1
}
func main() {
p := parallel.NewParallel()
p.Register(exceptionJob)
// miss the last argument on purpose
p.Except(exceptionHandler, "topic1")
p.Run()
}
more examples
Documentation
¶
Index ¶
- Variables
- type Handler
- type Parallel
- func (p *Parallel) Add(pipes ...*Pipeline) *Parallel
- func (p *Parallel) AddChildren(children ...*Parallel) *Parallel
- func (p *Parallel) Except(f interface{}, args ...interface{}) *Handler
- func (p *Parallel) NewChild() *Parallel
- func (p *Parallel) NewPipeline() *Pipeline
- func (p *Parallel) Register(f interface{}, args ...interface{}) *Handler
- func (p *Parallel) Run()
- func (p *Parallel) RunWithTimeOut(d time.Duration)
- type Pipeline
Constants ¶
This section is empty.
Variables ¶
var ( ErrArgNotFunction = errors.New("argument type not function") //argument type not function ErrInArgLenNotMatch = errors.New("input arguments length not match") //input arguments length not match ErrOutArgLenNotMatch = errors.New("output arguments length not match") //output arguments length not match ErrRecvArgTypeNotPtr = errors.New("receiver argument type is not pointer") //receiver argument type is not pointer ErrRecvArgNil = errors.New("receiver argument must not be nil") //receiver argument must not be nil )
Functions ¶
This section is empty.
Types ¶
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler instance
func NewHandler ¶
func NewHandler(f interface{}, args ...interface{}) *Handler
NewHandler create a new Handler which contains a single function call
func (*Handler) OnExcept ¶
func (h *Handler) OnExcept(e interface{})
OnExcept will executed by parallel when application panic occur Note that the type of e is unknown.
func (*Handler) SetReceivers ¶
SetReceivers sets the receivers of return values
type Parallel ¶
type Parallel struct {
// contains filtered or unexported fields
}
Parallel instance, which executes pipelines by parallel
func (*Parallel) AddChildren ¶
AddChildren add children to parallel to handle dependency
func (*Parallel) Except ¶
Except set the exception handling routine, when unexpected panic occur this routine will be executed.
func (*Parallel) NewPipeline ¶
NewPipeline create a new pipeline of parallel
func (*Parallel) RunWithTimeOut ¶
RunWithTimeOut start up all the jobs, and time out after d duration
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline instance, which executes jobs by serial
Source Files
Directories