queue_delay

Documentation

Index

• type DelayQueue
  • func NewDelayQueue(client cache.ICache, logger glog.ILogger) *DelayQueue
  • func (r *DelayQueue) GetDelayQueue(topic string) *RedisDelayQueue
• type RedisDelayQueue
  • func New(client cache.ICache, key string, logger glog.ILogger) *RedisDelayQueue
  • func (r *RedisDelayQueue) Acknowledge(keys ...string) int64
  • func (r *RedisDelayQueue) Add(value interface{}, delay int64) int64
  • func (r *RedisDelayQueue) Adds(values ...interface{}) int64
  • func (r *RedisDelayQueue) Count() int64
  • func (r *RedisDelayQueue) IsEmpty() bool
  • func (r *RedisDelayQueue) Remove(value ...interface{}) int64
  • func (r *RedisDelayQueue) Take(count int64) []string
  • func (r *RedisDelayQueue) TakeOne(timeout ...int64) string
  • func (r *RedisDelayQueue) TakeOneBlock(ctx context.Context, timeout ...int64) string
  • func (r *RedisDelayQueue) TransferAsync(ctx context.Context)
  • func (r *RedisDelayQueue) TryPop(value interface{}) bool

Types

type DelayQueue

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

func NewDelayQueue

func NewDelayQueue(client cache.ICache, logger glog.ILogger) *DelayQueue

func (*DelayQueue) GetDelayQueue

func (r *DelayQueue) GetDelayQueue(topic string) *RedisDelayQueue

type RedisDelayQueue

type RedisDelayQueue struct {
	DB int // redis DB 默认为 0

	Topic                       string // 消息队列主题
	ThrowOnFailure              bool   // 失败时抛出异常。默认false
	RetryTimesWhenSendFailed    int    // 发送消息失败时的重试次数。默认3次
	RetryIntervalWhenSendFailed int    // 重试间隔。默认1000ms
	TransferInterval            int64  // 转移延迟消息到主队列的间隔。默认10s
	Delay                       int64  // 默认延迟时间。默认60秒
	// contains filtered or unexported fields
}

func New

func New(client cache.ICache, key string, logger glog.ILogger) *RedisDelayQueue

func (*RedisDelayQueue) Acknowledge

func (r *RedisDelayQueue) Acknowledge(keys ...string) int64

Acknowledge 确认删除

func (*RedisDelayQueue) Add

func (r *RedisDelayQueue) Add(value interface{}, delay int64) int64

Add 添加延迟消息

func (*RedisDelayQueue) Adds

func (r *RedisDelayQueue) Adds(values ...interface{}) int64

Adds 批量生产

func (*RedisDelayQueue) Count

func (r *RedisDelayQueue) Count() int64

Count 个数

func (*RedisDelayQueue) IsEmpty

func (r *RedisDelayQueue) IsEmpty() bool

IsEmpty 集合是否为空

func (*RedisDelayQueue) Remove

func (r *RedisDelayQueue) Remove(value ...interface{}) int64

Remove 删除项

func (*RedisDelayQueue) Take

func (r *RedisDelayQueue) Take(count int64) []string

Take 获取一批

func (*RedisDelayQueue) TakeOne

func (r *RedisDelayQueue) TakeOne(timeout ...int64) string

TakeOne timeout 超时时间，默认0秒永远阻塞；负数表示直接返回，不阻塞。获取一个

func (*RedisDelayQueue) TakeOneBlock

func (r *RedisDelayQueue) TakeOneBlock(ctx context.Context, timeout ...int64) string

TakeOneBlock timeout 超时时间，默认0秒永远阻塞；负数表示直接返回，不阻塞。异步获取一个

func (*RedisDelayQueue) TransferAsync

func (r *RedisDelayQueue) TransferAsync(ctx context.Context)

func (*RedisDelayQueue) TryPop

func (r *RedisDelayQueue) TryPop(value interface{}) bool

TryPop 争夺消费，只有一个线程能够成功删除，