delayqueue

Documentation

Overview

基于rabbitmq的重试

Index

• type DelayMessage
• type DelayQueue
  • func New(logger log.Logger, amqpUrl string, exchangeName, queueName string, opt *Option, ...) (*DelayQueue, error)
  • func (r *DelayQueue) Publish(delayMsg *DelayMessage) error
  • func (r *DelayQueue) Shutdown()
• type Message
• type Option

Types

type DelayMessage

type DelayMessage struct {
	Body        []byte                 `json:"body"`        // 消息载体
	TotalTimes  int                    `json:"totalTimes"`  // 总重试次数
	ContentType string                 `json:"contentType"` // 可为空
	Metadata    map[string]interface{} `json:"metadata"`    // 附加信息
}

type DelayQueue

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

func New

func New(logger log.Logger, amqpUrl string, exchangeName, queueName string, opt *Option, handler func(msg *Message) error) (*DelayQueue, error)

func (*DelayQueue) Publish

func (r *DelayQueue) Publish(delayMsg *DelayMessage) error

func (*DelayQueue) Shutdown

func (r *DelayQueue) Shutdown()

type Message

type Message struct {
	*DelayMessage
	Times         int       `json:"times"`         // 当前重试次数
	CreateAt      time.Time `json:"createAt"`      // 首次发送时间
	LastPublishAt time.Time `json:"lastPublishAt"` // 上次发送时间
	TraceID       string    `json:"traceID"`       // 每次请求的TraceID
}

type Option

type Option struct {
	Concurrent  int            // 并发数量 默认为1
	BackOff     backoff.Policy // 默认 noPolicy
	ConsumerTag string         // 消费者标识
}