rabbitmq

Documentation

Index

• Constants
• type AmqpConfig
• type AmqpManager
  • func (self *AmqpManager) Client(dsname ...string) (*AmqpManager, error)
  • func (self *AmqpManager) InitConfig(input ...AmqpConfig)
  • func (self *AmqpManager) Publish(data MsgData, dlx ...DLX) error
  • func (self *AmqpManager) Pull(data LisData, callback func(msg MsgData) (MsgData, error)) (err error)
• type DLX
• type LisData
• type MQErrorLog
• type MsgData
• type PublishMQ
• type PublishManager
  • func (self *PublishManager) Client(dsname ...string) (*PublishManager, error)
  • func (self *PublishManager) GetChannel(data MsgData) (*QueueData, error)
  • func (self *PublishManager) InitConfig(input ...AmqpConfig) *PublishManager
  • func (self *PublishManager) Publish(data MsgData) error
• type PullManager
  • func (self *PullManager) AddPullReceiver(receivers ...*PullReceiver)
  • func (self *PullManager) Client(dsname ...string) (*PullManager, error)
  • func (self *PullManager) InitConfig(input ...AmqpConfig) *PullManager
• type PullReceiver
  • func (self *PullReceiver) Channel() *amqp.Channel
  • func (self *PullReceiver) ExchangeName() string
  • func (self *PullReceiver) OnError(err error)
  • func (self *PullReceiver) OnReceive(b []byte) bool
  • func (self *PullReceiver) QueueName() string
• type QueueData

Constants

const (
	MASTER        = "MASTER"
	DIRECT        = "direct"
	PrefetchCount = 50
)

Types

type AmqpConfig

type AmqpConfig struct {
	DsName   string
	Host     string
	Port     int
	Username string
	Password string
}

Amqp配置参数

type AmqpManager

type AmqpManager struct {
	DsName string
	// contains filtered or unexported fields
}

func (*AmqpManager) Client

func (self *AmqpManager) Client(dsname ...string) (*AmqpManager, error)

func (*AmqpManager) InitConfig

func (self *AmqpManager) InitConfig(input ...AmqpConfig)

func (*AmqpManager) Publish

func (self *AmqpManager) Publish(data MsgData, dlx ...DLX) error

根据通道发送信息,如通道不存在则自动创建

func (*AmqpManager) Pull

func (self *AmqpManager) Pull(data LisData, callback func(msg MsgData) (MsgData, error)) (err error)

监听指定队列消息

type DLX

type DLX struct {
	DlxExchange string                                 // 死信交换机
	DlxQueue    string                                 // 死信队列
	DlkExchange string                                 // 重读交换机
	DlkQueue    string                                 // 重读队列
	DlkCallFunc func(message MsgData) (MsgData, error) // 回调函数
}

Amqp延迟发送配置

type LisData

type LisData struct {
	Exchange      string
	Queue         string
	Kind          string
	PrefetchCount int
	PrefetchSize  int
	SendMgo       bool
	IsNack        bool
}

Amqp监听配置参数

type MQErrorLog

type MQErrorLog struct {
	Id       int64       `json:"id" bson:"_id" tb:"mq_error_log" mg:"true"`
	Exchange string      `json:"exchange" bson:"exchange"`
	Queue    string      `json:"queue" bson:"queue"`
	Content  interface{} `json:"content" bson:"content"`
	Type     int64       `json:"type" bson:"type"`
	Delay    int64       `json:"delay" bson:"delay"`
	Retries  int64       `json:"retries" bson:"retries"`
	Error    string      `json:"error" bson:"error"`
	Ctime    int64       `json:"ctime" bson:"ctime"`
	Utime    int64       `json:"utime" bson:"utime"`
	State    int64       `json:"state" bson:"state"`
}

Amqp消息异常日志

type MsgData

type MsgData struct {
	Exchange  string      `json:"exchange"`
	Queue     string      `json:"queue"`
	Kind      string      `json:"kind"`
	Content   interface{} `json:"content"`
	Type      int64       `json:"type"`
	Delay     int64       `json:"delay"`
	Retries   int64       `json:"retries"`
	Signature string      `json:"signature"`
}

Amqp消息参数

type PublishMQ

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

type PublishManager

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

func (*PublishManager) Client

func (self *PublishManager) Client(dsname ...string) (*PublishManager, error)

func (*PublishManager) GetChannel

func (self *PublishManager) GetChannel(data MsgData) (*QueueData, error)

客户端数 - 通道消息数

func (*PublishManager) InitConfig

func (self *PublishManager) InitConfig(input ...AmqpConfig) *PublishManager

func (*PublishManager) Publish

func (self *PublishManager) Publish(data MsgData) error

type PullManager

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

func (*PullManager) AddPullReceiver

func (self *PullManager) AddPullReceiver(receivers ...*PullReceiver)

func (*PullManager) Client

func (self *PullManager) Client(dsname ...string) (*PullManager, error)

func (*PullManager) InitConfig

func (self *PullManager) InitConfig(input ...AmqpConfig) *PullManager

type PullReceiver

type PullReceiver struct {
	Exchange string
	Queue    string
	LisData  LisData
	Callback func(msg MsgData) (MsgData, error)
	// contains filtered or unexported fields
}

监听对象

func (*PullReceiver) Channel

func (self *PullReceiver) Channel() *amqp.Channel

func (*PullReceiver) ExchangeName

func (self *PullReceiver) ExchangeName() string

func (*PullReceiver) OnError

func (self *PullReceiver) OnError(err error)

func (*PullReceiver) OnReceive

func (self *PullReceiver) OnReceive(b []byte) bool

func (*PullReceiver) QueueName

func (self *PullReceiver) QueueName() string

type QueueData

type QueueData struct {
	Name      string
	Messages  int
	Consumers int
}