mq

Documentation

Index

• Variables
• type Config
• type Consumer
  • func (c *Consumer) Start()
  • func (c *Consumer) Stop()
• type ConsumerWorker
• type ExchangeKind
• type Producer
  • func (producer *Producer) Destroy()
  • func (producer *Producer) Publish(msg interface{}) (err error)
  • func (producer *Producer) PurgeQueue() error
• type RabbitMqConsumer
  • func NewConsumer(ctx context.Context, worker ConsumerWorker, config *Config) RabbitMqConsumer
• type RabbitMqProducer
  • func NewMqProducer(config *Config) (RabbitMqProducer, error)

Variables

var ErrShouldDrop = errors.New("unprocessed message")

ErrShouldDrop 如果接收到的消息 consumer 无法处理，希望从队列中删除， 需要返回这个错误

Types

type Config

type Config struct {
	Addr          string
	Exchange      string
	ExchangeType  ExchangeKind // topic, direct, etc
	Queue         string
	RoutingKey    string
	ConsumerTag   string
	PrefetchCount int
	PrefetchSize  int
	ExchangeArgs  amqp.Table
	QueueArgs     amqp.Table
	QueueBindArgs amqp.Table
}

type Consumer

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

MQConsumer mq consumer 对象

func (*Consumer) Start

func (c *Consumer) Start()

Start 启动 mq consumer

func (*Consumer) Stop

func (c *Consumer) Stop()

Stop 关闭 consumer

type ConsumerWorker

type ConsumerWorker interface {
	Consume(context.Context, []byte) error
}

ConsumerWorker 处理从 MQ 得到的消息

type ExchangeKind

type ExchangeKind string

const (
	ExchangeTopic    ExchangeKind = amqp.ExchangeTopic
	ExchangeTFanout  ExchangeKind = amqp.ExchangeFanout
	ExchangeTDirect  ExchangeKind = amqp.ExchangeDirect
	ExchangeTHeaders ExchangeKind = amqp.ExchangeHeaders
)

type Producer

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

func (*Producer) Destroy

func (producer *Producer) Destroy()

func (*Producer) Publish

func (producer *Producer) Publish(msg interface{}) (err error)

func (*Producer) PurgeQueue

func (producer *Producer) PurgeQueue() error

PurgeQueue will purge all undelivered message of queue which declare in Config struct

type RabbitMqConsumer

type RabbitMqConsumer interface {
	Start()
	Stop()
}

func NewConsumer

func NewConsumer(ctx context.Context, worker ConsumerWorker, config *Config) RabbitMqConsumer

NewConsumer 创建一个 MQConsumer 实例

type RabbitMqProducer

type RabbitMqProducer interface {
	Destroy()
	Publish(interface{}) error
	PurgeQueue() error
}

func NewMqProducer

func NewMqProducer(config *Config) (RabbitMqProducer, error)