rabbitmq

Documentation

Index

• func ConnectRabbitMQ(conf AmqpConfig) (*amqp.Connection, error)
• type AmqpConfig
• type Config
• type DLX
• type MsgData
• type Option
• type PublishMQ
• type PublishManager
  • func NewPublish(ds ...string) (*PublishManager, error)
  • func (self *PublishManager) Client(ds ...string) (*PublishManager, error)
  • func (self *PublishManager) Connect() (*PublishManager, error)
  • func (self *PublishManager) InitConfig(input ...AmqpConfig) (*PublishManager, error)
  • func (self *PublishManager) Publish(exchange, queue string, dataType int64, content interface{}) error
  • func (self *PublishManager) PublishMsgData(data *MsgData) error
  • func (self *PublishManager) Queue(data *MsgData) (*QueueData, error)
• type PullManager
  • func NewPull(ds ...string) (*PullManager, error)
  • func (self *PullManager) AddPullReceiver(receivers ...*PullReceiver)
  • func (self *PullManager) Client(ds ...string) (*PullManager, error)
  • func (self *PullManager) Connect() (*PullManager, error)
  • func (self *PullManager) InitConfig(input ...AmqpConfig) (*PullManager, error)
• type PullReceiver
  • func (self *PullReceiver) OnError(err error)
  • func (self *PullReceiver) OnReceive(b []byte) bool
• type QueueData

Functions

func ConnectRabbitMQ

func ConnectRabbitMQ(conf AmqpConfig) (*amqp.Connection, error)

Types

type AmqpConfig

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

type Config

type Config struct {
	Option        Option
	Durable       bool
	PrefetchCount int
	PrefetchSize  int
	IsNack        bool
	AutoAck       bool
}

Amqp监听配置参数

type DLX

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

type MsgData

type MsgData struct {
	Option    Option      `json:"op"`
	Durable   bool        `json:"du"`
	Content   interface{} `json:"co"`
	Type      int64       `json:"ty"`
	Delay     int64       `json:"dy"`
	Retries   int64       `json:"rt"`
	Nonce     string      `json:"no"`
	Signature string      `json:"sg"`
}

type Option

type Option struct {
	Exchange string `json:"ex"`
	Queue    string `json:"qe"`
	Kind     string `json:"kd"`
	Router   string `json:"ru"`
	SigTyp   int    `json:"st"` // 是否加密 0.明文签名 1.AES加密签名
	SigKey   string `json:"-"`  // 验签密钥
}

type PublishMQ

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

type PublishManager

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

func NewPublish

func NewPublish(ds ...string) (*PublishManager, error)

func (*PublishManager) Client

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

func (*PublishManager) Connect

func (self *PublishManager) Connect() (*PublishManager, error)

func (*PublishManager) InitConfig

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

func (*PublishManager) Publish

func (self *PublishManager) Publish(exchange, queue string, dataType int64, content interface{}) error

func (*PublishManager) PublishMsgData

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

func (*PublishManager) Queue

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

type PullManager

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

func NewPull

func NewPull(ds ...string) (*PullManager, error)

func (*PullManager) AddPullReceiver

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

func (*PullManager) Client

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

func (*PullManager) Connect

func (self *PullManager) Connect() (*PullManager, error)

func (*PullManager) InitConfig

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

type PullReceiver

type PullReceiver struct {
	Config       *Config
	ContentInter func(typ int64) interface{}
	Callback     func(msg *MsgData) error
	Debug        bool // 是否打印具体pull数据实体
	Delay        int  // pull失败重试间隔
	// contains filtered or unexported fields
}

func (*PullReceiver) OnError

func (self *PullReceiver) OnError(err error)

func (*PullReceiver) OnReceive

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

type QueueData

type QueueData struct {
	Name      string
	Consumers int
	Messages  int
}