grpool

README

pkg说明

grpool

在gapi框架中不建议使用go goroutine开go协程，因为如果go协程如果没有recover()的话，程序panic会无法捕获。 要求使用此pkg下的grpool.Submit(task func())来运行go协程，此方法自带了recover()，可以捕获panic。

TaskGroup

串行或并发地执行一组任务

type TaskE = func(ctx context.Context) error

// RunTask serial run all tasks, one after the other.
// if any task has return error, other tasks will not run
// return the error.
func RunTask(ctx context.Context, tasks ...TaskE) error 

// CRunTask concurrence run all tasks.
// if any task has return error, all tasks will be killed (Don't take it too seriously)
// return the first error.
func CRunTask(ctx context.Context, tasks ...TaskE) error 

// CRunTaskE concurrence run all tasks.
// if any task has return error, all tasks will not be killed (Don't take it too seriously)
// return all errors.
func CRunTaskE(ctx context.Context, tasks ...TaskE) (timeout bool, errs error) 

`CRunTask`
所有任务将被并发执行（无序），如果任何一个任务`发生错误`或`ctx超时` ，`CRunTaskE` 将返回第一个错误。

`CRunTaskE`
所有任务将被并发执行（无序），任何一个任务`发生错误`都不会影响其他任务，并收集所有任务的error。
但如果是ctx超时，`CRunTaskE`将不再等待未完成的任务直接返回。

`RunTask`
所有任务按照`FIFO`被串行执行，如果任何一个任务在执行过程中`发生错误`或`ctx超时`，则剩余任务将不再执行，返回当前错误。

Documentation

Index

• Constants
• func CRunTask(ctx context.Context, tasks ...TaskE) error
• func CRunTaskE(ctx context.Context, tasks ...TaskE) (timeout bool, errs error)
• func Cap() int
• func ErrWrap(old, new error) error
• func Free() int
• func Release() error
• func RunTask(ctx context.Context, tasks ...TaskE) error
• func Running() int
• func SetLog(grpoolLog Logger)
• func Submit(task func()) error
• func Tune(size int)
• type DispatchType
• type GrFuncPool
  • func NewPoolWithFunc(size int, pf func(interface{})) (*GrFuncPool, error)
  • func NewTimingPoolWithFunc(size int, expiry time.Duration, pf func(interface{})) (*GrFuncPool, error)
• type GrPool
  • func NewPool(size int) (*GrPool, error)
  • func NewPoolWithExpire(size int, expiry time.Duration) (*GrPool, error)
  • func (p *GrPool) Cap() int
  • func (p *GrPool) Free() int
  • func (p *GrPool) Release() error
  • func (p *GrPool) Running() int
  • func (p *GrPool) Submit(task func()) error
  • func (p *GrPool) Tune(size int)
• type Logger
• type TaskE
• type TaskGroup
  • func NewTaskGroup() *TaskGroup
  • func (*TaskGroup) CRunTask(pool *GrPool, ctx context.Context, tasks ...TaskE) error
  • func (*TaskGroup) CRunTaskE(pool *GrPool, ctx context.Context, tasks ...TaskE) (timeout bool, errs error)
  • func (*TaskGroup) RunTask(pool *GrPool, ctx context.Context, tasks ...TaskE) error

Constants

const (
	BreakByError = true
	NoBreak      = false

	SerialDispatch DispatchType = iota
	ConcurrentDispatch
)

Functions

func CRunTask

func CRunTask(ctx context.Context, tasks ...TaskE) error

CRunTask concurrence run all tasks. if any task has return error, all tasks will be killed (Don't take it too seriously) return the first error.

func CRunTaskE

func CRunTaskE(ctx context.Context, tasks ...TaskE) (timeout bool, errs error)

CRunTaskE concurrence run all tasks. if any task has return error, all tasks will not be killed (Don't take it too seriously) return all errors.

func Cap

func Cap() int

func ErrWrap

func ErrWrap(old, new error) error

func Free

func Free() int

func Release

func Release() error

func RunTask

func RunTask(ctx context.Context, tasks ...TaskE) error

RunTask serial run all tasks, one after the other. if any task has return error, other tasks will not run return the first error.

func Running

func Running() int

func SetLog

func SetLog(grpoolLog Logger)

func Submit

func Submit(task func()) error

func Tune

func Tune(size int)

Types

type DispatchType

type DispatchType int

type GrFuncPool

type GrFuncPool struct {
	*ants.PoolWithFunc
}

============================================================================= pool with function

func NewPoolWithFunc

func NewPoolWithFunc(size int, pf func(interface{})) (*GrFuncPool, error)

Pool with a argument

func NewTimingPoolWithFunc

func NewTimingPoolWithFunc(size int, expiry time.Duration, pf func(interface{})) (*GrFuncPool, error)

type GrPool

type GrPool struct {
	// contains filtered or unexported fields
}

func NewPool

func NewPool(size int) (*GrPool, error)

func NewPoolWithExpire

func NewPoolWithExpire(size int, expiry time.Duration) (*GrPool, error)

func (*GrPool) Cap

func (p *GrPool) Cap() int

func (*GrPool) Free

func (p *GrPool) Free() int

func (*GrPool) Release

func (p *GrPool) Release() error

func (*GrPool) Running

func (p *GrPool) Running() int

func (*GrPool) Submit

func (p *GrPool) Submit(task func()) error

func (*GrPool) Tune

func (p *GrPool) Tune(size int)

type Logger

type Logger interface {
	Debugf(template string, args ...interface{})
	Infof(template string, args ...interface{})
	Warnf(template string, args ...interface{})
	Errorf(template string, args ...interface{})
}

type TaskE

type TaskE func(ctx context.Context) error

type TaskGroup

type TaskGroup struct{}

func NewTaskGroup

func NewTaskGroup() *TaskGroup

func (*TaskGroup) CRunTask

func (*TaskGroup) CRunTask(pool *GrPool, ctx context.Context, tasks ...TaskE) error

func (*TaskGroup) CRunTaskE

func (*TaskGroup) CRunTaskE(pool *GrPool, ctx context.Context, tasks ...TaskE) (timeout bool, errs error)

func (*TaskGroup) RunTask

func (*TaskGroup) RunTask(pool *GrPool, ctx context.Context, tasks ...TaskE) error