delayqueue

README

delay-queue

基于ouqiang/delay-queue，将核心的延时队列逻辑抽离了出来

Documentation

Index

• Constants
• type BucketItem
• type DelayRedisQueue
  • func New(ctx context.Context, name string, bucketCnt int, redisClient *redis.Client) *DelayRedisQueue
  • func (q *DelayRedisQueue) Get(ctx context.Context, jobId string) (*Job, error)
  • func (q *DelayRedisQueue) Pop(ctx context.Context, topics []string) (*Job, error)
  • func (q *DelayRedisQueue) Push(ctx context.Context, job Job) error
  • func (q *DelayRedisQueue) Remove(ctx context.Context, jobId string) error
• type Job

Constants

const (
	DefaultQueueBlockTimeout = 10 * time.Second
)

Types

type BucketItem

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

BucketItem bucket中的元素

type DelayRedisQueue

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

func New

func New(ctx context.Context, name string, bucketCnt int, redisClient *redis.Client) *DelayRedisQueue

func (*DelayRedisQueue) Get

func (q *DelayRedisQueue) Get(ctx context.Context, jobId string) (*Job, error)

Get 查询Job

func (*DelayRedisQueue) Pop

func (q *DelayRedisQueue) Pop(ctx context.Context, topics []string) (*Job, error)

Pop 轮询获取Job

func (*DelayRedisQueue) Push

func (q *DelayRedisQueue) Push(ctx context.Context, job Job) error

Push 添加一个Job到队列中

func (*DelayRedisQueue) Remove

func (q *DelayRedisQueue) Remove(ctx context.Context, jobId string) error

Remove 删除Job

type Job

type Job struct {
	Topic string `json:"topic" msgpack:"1"`
	Id    string `json:"id" msgpack:"2"`    // job唯一标识ID
	Delay int64  `json:"delay" msgpack:"3"` // 延迟时间, 单位为s
	TTR   int64  `json:"ttr" msgpack:"4"`   // Time-To-Retry，失败后，多久再次重试
	Body  string `json:"body" msgpack:"5"`
}

Job 使用msgpack序列化后保存到Redis,减少内存占用