rabbitmq

Documentation

Index

• func New(config map[string]interface{}) (interface{}, error)
• type ChannelBox
  • func (c *ChannelBox) Close(force bool) error
  • func (c *ChannelBox) GetChannel() *amqp.Channel
• type Client
  • func (c *Client) Consume(parameter *ConsumeData) (<-chan amqp.Delivery, error)
  • func (c *Client) DecodeBody(d amqp.Delivery, ret interface{}) error
  • func (c *Client) DecodeHeaders(d amqp.Delivery) *RabbitHeaders
  • func (c *Client) FreeChannel() (*ChannelBox, error)
  • func (c *Client) GetConsumeChannelBox(queueName string, opCodes []string, exchange *ExchangeData) (*ChannelBox, error)
  • func (c *Client) Publish(parameter *PublishData, logId string) (bool, error)
  • func (c *Client) SetExchangeDeclare(exchange *ExchangeData) error
• type ConnBox
  • func (c *ConnBox) Disable() bool
• type ConsumeData
• type ExchangeData
• type Pool
  • func (c *Pool) ExchangeType(exchangeType string) string
  • func (c *Pool) GetServers() (servers []string)
  • func (c *Pool) Init() error
  • func (c *Pool) ServiceName(serviceName string) string
  • func (c *Pool) SetExchangeName(v string)
  • func (c *Pool) SetExchangeType(v string)
  • func (c *Pool) SetLogger(logger logs.ILogger)
  • func (c *Pool) SetMaxChannelNum(v int)
  • func (c *Pool) SetMaxIdleChannel(v int)
  • func (c *Pool) SetMaxWaitTime(v string) error
  • func (c *Pool) SetPass(v string)
  • func (c *Pool) SetProbeInterval(v string) error
  • func (c *Pool) SetServers(v []interface{})
  • func (c *Pool) SetServiceName(v string)
  • func (c *Pool) SetTlsCert(v string)
  • func (c *Pool) SetTlsCertKey(v string)
  • func (c *Pool) SetTlsRootCAs(v string)
  • func (c *Pool) SetUser(v string)
• type PublishData
• type RabbitHeaders

Functions

func New

func New(config map[string]interface{}) (interface{}, error)

RabbitMq client component, support Publisher-Consumer configuration: components:

rabbitMq:
    tlsRootCAs:""
    tlsCert: ""
    tlsCertKey: ""
    user: "guest"
    pass: "guest"
    exchangeName: ""
    exchangeType: ""
    maxChannelNum: 2000
    maxIdleChannel: "200"
    maxIdleChannelTime:"10s"
    probeInterval: "0s"
    maxWaitTime: "200ms"
    serviceName: "pgo-xxx"
    servers:
        - "127.0.0.1:6379"
        - "127.0.0.1:6380"

Types

type ChannelBox

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

func (*ChannelBox) Close

func (c *ChannelBox) Close(force bool) error

func (*ChannelBox) GetChannel

func (c *ChannelBox) GetChannel() *amqp.Channel

type Client

type Client struct {
	Pool
}

func (*Client) Consume

func (c *Client) Consume(parameter *ConsumeData) (<-chan amqp.Delivery, error)

func (*Client) DecodeBody

func (c *Client) DecodeBody(d amqp.Delivery, ret interface{}) error

func (*Client) DecodeHeaders

func (c *Client) DecodeHeaders(d amqp.Delivery) *RabbitHeaders

func (*Client) FreeChannel

func (c *Client) FreeChannel() (*ChannelBox, error)

func (*Client) GetConsumeChannelBox

func (c *Client) GetConsumeChannelBox(queueName string, opCodes []string, exchange *ExchangeData) (*ChannelBox, error)

func (*Client) Publish

func (c *Client) Publish(parameter *PublishData, logId string) (bool, error)

func (*Client) SetExchangeDeclare

func (c *Client) SetExchangeDeclare(exchange *ExchangeData) error

type ConnBox

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

func (*ConnBox) Disable

func (c *ConnBox) Disable() bool

type ConsumeData

type ConsumeData struct {
	ExChange  *ExchangeData
	QueueName string
	Name      string
	OpCodes   []string
	AutoAck   bool
	NoWait    bool
	Exclusive bool
	Limit     int
}

type ExchangeData

type ExchangeData struct {
	Name       string // 交换机名
	Type       string // 交换机类型
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Args       amqp.Table
}

type Pool

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

func (*Pool) ExchangeType

func (c *Pool) ExchangeType(exchangeType string) string

func (*Pool) GetServers

func (c *Pool) GetServers() (servers []string)

func (*Pool) Init

func (c *Pool) Init() error

func (*Pool) ServiceName

func (c *Pool) ServiceName(serviceName string) string

func (*Pool) SetExchangeName

func (c *Pool) SetExchangeName(v string)

func (*Pool) SetExchangeType

func (c *Pool) SetExchangeType(v string)

func (*Pool) SetLogger

func (c *Pool) SetLogger(logger logs.ILogger)

func (*Pool) SetMaxChannelNum

func (c *Pool) SetMaxChannelNum(v int)

func (*Pool) SetMaxIdleChannel

func (c *Pool) SetMaxIdleChannel(v int)

func (*Pool) SetMaxWaitTime

func (c *Pool) SetMaxWaitTime(v string) error

func (*Pool) SetPass

func (c *Pool) SetPass(v string)

func (*Pool) SetProbeInterval

func (c *Pool) SetProbeInterval(v string) error

func (*Pool) SetServers

func (c *Pool) SetServers(v []interface{})

func (*Pool) SetServiceName

func (c *Pool) SetServiceName(v string)

func (*Pool) SetTlsCert

func (c *Pool) SetTlsCert(v string)

func (*Pool) SetTlsCertKey

func (c *Pool) SetTlsCertKey(v string)

func (*Pool) SetTlsRootCAs

func (c *Pool) SetTlsRootCAs(v string)

func (*Pool) SetUser

func (c *Pool) SetUser(v string)

type PublishData

type PublishData struct {
	DeliveryMode uint8  // Transient (0 or 1)  or Persistent (2)
	ServiceName  string // 服务名
	ExChange     *ExchangeData
	OpCode       string      // 操作code 和queue绑定相关
	OpUid        string      // 操作用户id 可以为空
	ContentType  string      // 内容类型 默认为："text/plain"
	Data         interface{} // 发送数据
}

rabbit 发布结构

type RabbitHeaders

type RabbitHeaders struct {
	LogId     string
	Exchange  string
	RouteKey  string
	Service   string
	OpUid     string
	Timestamp time.Time
	MessageId string
}