butcher

README

butcher

Overview

Butcher is a library providing a simple way to execute some task concurrency. Integrates convenient features such as concurrency control and retry.

Quick Start

package main

import (
    "context"
    "fmt"
    "github.com/KawashiroNitori/butcher"
)

type executor struct{}

func (e *executor) GenerateJob(ctx context.Context, jobCh chan<- interface{}) error {
    // generate your jobs here
    for i := 0; i < 100; i++ {
        // you can check canceled context or not, is up to you
        select {
        case <- ctx.Done():
            return nil
        default:
        }
        
        // push your job into channel
        jobCh <- i
    }
    // you may NOT close jobCh manually
    return nil
}

func (e *executor) Task(ctx context.Context, job interface{}) error {
    // execute your job here
    fmt.Printf("job %v finished!\n", job)
    return nil
}

// OnFinish implement an optional func to check your job is finished if you want
func (e *executor) OnFinish(ctx context.Context, job interface{}, err error) {
    if err != nil {
        fmt.Println("job %v error: %v", job, err)
    }
}

func main() {
    ctx := context.Background()
    b, err := butcher.NewButcher(&executor{})  // you can add some options here
    if err != nil {
        panic(err)
    }
    
    err = b.Run(ctx)
    if err != nil {
        panic(err)
    }
}

Documentation

Overview

Example (NewButcher_Options)

ctx := context.Background()
executor := &basicExecutor{}
b, err := NewButcher(
	executor,
	MaxWorker(3),               // specifies maximum number of concurrent workers, default is 1.
	BufferSize(20),             // specifies job buffer size, recommended value is bigger than MaxWorker and RateLimit, default is 1.
	RateLimit(10.0),            // control task execute speed, default is Infinity.
	TaskTimeout(1*time.Second), // specifies task execute timeout, returns a context.TimeExceeded error if timeout is exceeded, default no timeout.
	RetryOnError(3),            // specifies retry times when task return error, default no retry.
	InterruptSignal(syscall.SIGINT, syscall.SIGTERM), // specified signals can interrupt task running, default is syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGQUIT.
)
if err != nil {
	panic(err)
}
err = b.Run(ctx)
if err != nil {
	panic(err)
}
fmt.Println("all done")

Index

• type Butcher
  • func NewButcher(executor Executor, opts ...Option) (Butcher, error)
• type Executor
• type OnFinishWatcher
• type Option
  • func BufferSize(size int) Option
  • func InterruptSignal(signals ...os.Signal) Option
  • func MaxWorker(count int) Option
  • func RateLimit(tasksPerSecond float64) Option
  • func RetryOnError(maxTimes int) Option
  • func TaskTimeout(timeout time.Duration) Option

Examples

• Package (NewButcher_Options)

Types

type Butcher

type Butcher interface {
	Run(context.Context) error
}

func NewButcher

func NewButcher(executor Executor, opts ...Option) (Butcher, error)

NewButcher returns a butcher object for execute task executor. It has some options to control execute behaviors. if no options given, it runs tasks serially.

type Executor

type Executor interface {
	// GenerateJob generate your jobs here. put your job into jobCh, don't close jobCh manually.
	GenerateJob(ctx context.Context, jobCh chan<- interface{}) error
	// Task execute your job here. It will be scheduled by butcher.
	Task(ctx context.Context, job interface{}) error
}

Executor implements task executor interface.

type OnFinishWatcher

type OnFinishWatcher interface {
	OnFinish(ctx context.Context, job interface{}, err error)
}

OnFinishWatcher implements optional OnFinish function if you want to watch the result of job.

type Option

type Option func(b *butcher) error

func BufferSize

func BufferSize(size int) Option

BufferSize specifies job buffer size, recommended value is bigger than MaxWorker and RateLimit, default is 1.

func InterruptSignal

func InterruptSignal(signals ...os.Signal) Option

InterruptSignal specified signals can interrupt task running. Default is syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGQUIT.

func MaxWorker

func MaxWorker(count int) Option

MaxWorker specifies maximum number of concurrent workers, default is 1.

func RateLimit

func RateLimit(tasksPerSecond float64) Option

RateLimit control task execute speed, default is Infinity.

func RetryOnError

func RetryOnError(maxTimes int) Option

RetryOnError specifies retry times when task return error, default no retry.

func TaskTimeout

func TaskTimeout(timeout time.Duration) Option

TaskTimeout specifies task execute timeout, returns a context.TimeExceeded error if timeout is exceeded, default no timeout.