queue

Documentation

Index

• type Handler
• type Option
  • func Interval(s int64) Option
  • func Logger(s *zap.SugaredLogger) Option
  • func Step(s int64) Option
• type Options
  • func DefaultOptions() Options
• type Queue
  • func New(store Store, opts ...Option) *Queue
  • func (q *Queue) Load(handlers map[string]Handler) *Queue
  • func (q *Queue) Pop(t *Task) error
  • func (q *Queue) Push(t *Task) error
  • func (q *Queue) Register(tag string, h Handler)
• type Store
• type Task
  • func UnmarshalJson(j string) (*Task, error)
  • func (t *Task) MarshalJson() string
  • func (t *Task) Sn() uint64
  • func (t *Task) Subkey() string

Types

type Handler

type Handler func(*Task) error

任务执行函数。 特别注意，如果ttl不是0。但是因为某些原因不需要执行下一次操作了。则需要返回nil 比如。需要对某特定条件数据执行某个操作。在这之前自然是会检查这个数据是否存在或者是否能被操作等。 如果没有拿到这条数据，则直接返回nil

type Option

type Option func(o *Options)

Option Option

func Interval

func Interval(s int64) Option

func Logger

func Logger(s *zap.SugaredLogger) Option

func Step

func Step(s int64) Option

type Options

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

func DefaultOptions

func DefaultOptions() Options

DefaultOptions default options

type Queue

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

func New

func New(store Store, opts ...Option) *Queue

func (*Queue) Load

func (q *Queue) Load(handlers map[string]Handler) *Queue

Load 任务内容

func (*Queue) Pop

func (q *Queue) Pop(t *Task) error

func (*Queue) Push

func (q *Queue) Push(t *Task) error

func (*Queue) Register

func (q *Queue) Register(tag string, h Handler)

Register 执行方法

type Store

type Store interface {
	Push(t *Task) error   //添加任务
	Pop(t *Task) error    //剔除任务
	Update(t *Task) error //更新任务--重复调用Push其实就是update了。感觉这个有点多余
	//获取单个执行任务。
	ReadTask(key string) (*Task, error)

	//拉取所有任务队列.返回任务名称
	FetchJob(step int64) ([]string, error)
}

type Store struct {
	store StoreInterface
}

type Task

type Task struct {
	Key   string                 // system_name/job_id
	Tag   string                 // Tag匹配Handler，无Tag的Task将不会被执行
	Args  map[string]interface{} // 任务参数
	Delay int64                  // 延迟时间

	TTL      int64 //如果执行失败，下一次重复的时间
	Retry    int   //已经重试的次数
	RetryMax int   //最大重试次数
}

func UnmarshalJson

func UnmarshalJson(j string) (*Task, error)

func (*Task) MarshalJson

func (t *Task) MarshalJson() string

func (*Task) Sn

func (t *Task) Sn() uint64

把key变成sn。很多情况下，都是sn设为key的

func (*Task) Subkey

func (t *Task) Subkey() string