rabbit

Documentation

Index

• Constants
• Variables
• func Init(rbInfo *RbInfo)
• type ConsumeOpt
• type DialOption
  • func WithConsumeOpt(consumeOpt ConsumeOpt) DialOption
  • func WithDeadDurable(durable bool) DialOption
  • func WithDeadExchangeOpt(exchangeOpt ExchangeDeclareOpt) DialOption
  • func WithDeadQueueBindOpt(queueBindOpt QueueBindOpt) DialOption
  • func WithDeadQueueOpt(queueOpt QueueDeclareOpt) DialOption
  • func WithDurable(durable bool) DialOption
  • func WithExchangeOpt(exchangeOpt ExchangeDeclareOpt) DialOption
  • func WithQueueBindOpt(queueBindOpt QueueBindOpt) DialOption
  • func WithQueueOpt(queueOpt QueueDeclareOpt) DialOption
  • func WithReconnectDelay(reconnectDelay time.Duration) DialOption
  • func WithResendDelay(resendDelay time.Duration) DialOption
• type ExchangeDeclareOpt
• type QueueBindOpt
• type QueueDeclareOpt
• type RbInfo
• type RbMqClient
  • func NewAndInitRabbitClient(rbInfo RbInfo, callBack func(amqp.Delivery), opts ...DialOption) (*RbMqClient, error)
  • func NewAndInitRabbitServer(rbInfo RbInfo, opts ...DialOption) (*RbMqClient, error)
  • func NewRabbitProduct(rbInfo RbInfo, callBack func(amqp.Delivery), opts ...DialOption) (*RbMqClient, error)
  • func (rabbitProduct *RbMqClient) Close()
  • func (rabbitProduct *RbMqClient) Consume(prefetchCount int)
  • func (rabbitProduct *RbMqClient) InitRabbitProduct()
  • func (rabbitProduct *RbMqClient) Publish(msg []byte) bool
  • func (rabbitProduct *RbMqClient) PublishMulti(msgs [][]byte) (bool, int)

Constants

const (
	ReconnectDelay = 10 // 连接断开后多久重连
	ResendDelay    = 3  // 消息发送超时时间
)

Variables

var (
	ParamsError   = errors.New("地址、路由和队列不能为空")
	CallBackError = errors.New("消费回调函数不能为空")
)

Functions

func Init

func Init(rbInfo *RbInfo)

Types

type ConsumeOpt

type ConsumeOpt struct {
	NoLocal   bool
	NoAck     bool
	Exclusive bool
	NoWait    bool
	Arguments amqp.Table
}

type DialOption

type DialOption interface {
	// contains filtered or unexported methods
}

func WithConsumeOpt

func WithConsumeOpt(consumeOpt ConsumeOpt) DialOption

WithConsumeOpt Consume 参数

func WithDeadDurable

func WithDeadDurable(durable bool) DialOption

WithDeadDurable 死信队列是否持久化

func WithDeadExchangeOpt

func WithDeadExchangeOpt(exchangeOpt ExchangeDeclareOpt) DialOption

WithDeadExchangeOpt 死信交换机配置, 只有配置了死信队列才生效

func WithDeadQueueBindOpt

func WithDeadQueueBindOpt(queueBindOpt QueueBindOpt) DialOption

WithDeadQueueBindOpt 死信队列绑定配置, 只有配置了死信队列才生效

func WithDeadQueueOpt

func WithDeadQueueOpt(queueOpt QueueDeclareOpt) DialOption

WithDeadQueueOpt 死信队列配置, 只有配置了死信队列才生效

func WithDurable

func WithDurable(durable bool) DialOption

WithDurable 是否持久化

func WithExchangeOpt

func WithExchangeOpt(exchangeOpt ExchangeDeclareOpt) DialOption

WithExchangeOpt 交换机配置

func WithQueueBindOpt

func WithQueueBindOpt(queueBindOpt QueueBindOpt) DialOption

WithQueueBindOpt 队列绑定配置

func WithQueueOpt

func WithQueueOpt(queueOpt QueueDeclareOpt) DialOption

WithQueueOpt 队列配置

func WithReconnectDelay

func WithReconnectDelay(reconnectDelay time.Duration) DialOption

WithReconnectDelay 链接超时时间

func WithResendDelay

func WithResendDelay(resendDelay time.Duration) DialOption

WithResendDelay 发送超时时间

type ExchangeDeclareOpt

type ExchangeDeclareOpt struct {
	ExchangeType string
	AutoDelete   bool
	Internal     bool
	NoWait       bool
	Arguments    amqp.Table
}

type QueueBindOpt

type QueueBindOpt struct {
	NoWait    bool
	Arguments amqp.Table
}

type QueueDeclareOpt

type QueueDeclareOpt struct {
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
	Arguments  amqp.Table
}

type RbInfo

type RbInfo struct {
	Addr             string
	ExchangeName     string
	QueueName        string
	RouteKey         string
	DeadExchangeName string
	DeadQueueName    string
	DeadRouteKey     string
	// contains filtered or unexported fields
}

type RbMqClient

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

func NewAndInitRabbitClient

func NewAndInitRabbitClient(rbInfo RbInfo, callBack func(amqp.Delivery), opts ...DialOption) (*RbMqClient, error)

NewAndInitRabbitClient 新建消费端(消费消息)

func NewAndInitRabbitServer

func NewAndInitRabbitServer(rbInfo RbInfo, opts ...DialOption) (*RbMqClient, error)

NewAndInitRabbitServer 新建服务端(发布消息)

func NewRabbitProduct

func NewRabbitProduct(rbInfo RbInfo, callBack func(amqp.Delivery), opts ...DialOption) (*RbMqClient, error)

func (*RbMqClient) Close

func (rabbitProduct *RbMqClient) Close()

Close 关闭, 如果不关闭的话, channel 不会释放

func (*RbMqClient) Consume

func (rabbitProduct *RbMqClient) Consume(prefetchCount int)

Consume 消费消息 prefetchCount 预取数量，设置为1的话，可以实现性能高的服务器消费的数量多

func (*RbMqClient) InitRabbitProduct

func (rabbitProduct *RbMqClient) InitRabbitProduct()

func (*RbMqClient) Publish

func (rabbitProduct *RbMqClient) Publish(msg []byte) bool

Publish 发布消息

func (*RbMqClient) PublishMulti

func (rabbitProduct *RbMqClient) PublishMulti(msgs [][]byte) (bool, int)

PublishMulti 返回发送是否全部成功和发送成功的数量