executor

README

Executor

Executor is a simple worker pool implemented for Golang.

Features:

• Execute concurrent job by Goroutine
• Rate limiter
• Dynamic handler by reflect

More information at Blog.

Install

$ go get github.com/thinhdanggroup/executor

Usage

executor, err := executor.NewExecutor(executor.DefaultExecutorConfig())

if err != nil {
  logrus.Error(err)
}

// close resource before quit
defer executor.Close()

for i := 0; i < 10; i++ {
  executor.Publish(func(input int) {
    fmt.Printf("2 * %d = %d \n", input, 2*input)
  }, i)

  executor.Publish(func(input int) {
    fmt.Printf("2 ^ %d = %d \n", input, input^2)
  }, i)

  executor.Publish(func(a int, b int) {
    fmt.Printf("%d + %d = %d \n", a, b, a+b)
  }, i, i+1)
}

// Output:
// 2 * 0 = 0 
// 2 ^ 0 = 2 
// 2 ^ 1 = 3 
// 1 + 2 = 3 
// 2 * 2 = 4 
// 2 ^ 2 = 0 
// 2 + 3 = 5 
// 0 + 1 = 1 
// 2 * 1 = 2

Documentation

Index

• type Config
  • func DefaultConfig() Config
• type Executor
  • func New(config Config) (*Executor, error)
  • func (pipeline *Executor) Close()
  • func (pipeline *Executor) Publish(handler interface{}, inputArgs ...interface{}) error
  • func (pipeline *Executor) PublishJob(job *Job)
  • func (pipeline *Executor) Wait()
• type Job
  • func NewJob(handler interface{}, inputArgs ...interface{}) (*Job, error)

Types

type Config

type Config struct {
	ReqPerSeconds int
	QueueSize     int
	NumWorkers    int
}

Config is a config of executor. ReqPerSeconds is request per seconds. If it is 0, no limit for requests. QueueSize is size of buffer. Executor use synchronize channel, publisher will waiting if channel is full. NumWorkers is a number of goroutine.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig is a default config

type Executor

type Executor struct {
	RateLimit  ratelimit.Limiter
	WaitGroup  *sync.WaitGroup
	Channel    chan *Job
	NumWorkers int
}

Executor is a simple thread pool base on goroutine.

func New

func New(config Config) (*Executor, error)

New returns a Executors that will manage workers.

func (*Executor) Close

func (pipeline *Executor) Close()

Close channel and wait all worker done.

func (*Executor) Publish

func (pipeline *Executor) Publish(handler interface{}, inputArgs ...interface{}) error

Publish to publish a handler and arguments Workers will run handler with provided arguments.

func (*Executor) PublishJob

func (pipeline *Executor) PublishJob(job *Job)

PublishJob publish a provided job.

func (*Executor) Wait

func (pipeline *Executor) Wait()

Wait for all worker done.

type Job

type Job struct {
	Handler interface{}
	Args    []reflect.Value
}

Job is a task will be executor execute.

func NewJob

func NewJob(handler interface{}, inputArgs ...interface{}) (*Job, error)

NewJob returns a Job that will be executed in workers.