rabbitmq

Documentation

Index

• type ConnectionConfig
• type ConsumerConfig
• type ExchangeConfig
• type MessageConfig
• type MonitoringConfig
• type Options
• type PoolConfig
• type ProducerConfig
• type QueueConfig
• type RabbitMQClient
  • func NewRabbitMQ(opt *Options, l log.Logger) *RabbitMQClient
  • func (r *RabbitMQClient) BindQueue(queueName, routingKey, exchangeName string) error
  • func (r *RabbitMQClient) Connect() error
  • func (r *RabbitMQClient) Consume(queueName string, handler func(amqp.Delivery)) error
  • func (r *RabbitMQClient) DeclareExchange(name, kind string) error
  • func (r *RabbitMQClient) DeclareQueue(name string) (amqp.Queue, error)
  • func (r *RabbitMQClient) Disconnect() error
  • func (r *RabbitMQClient) GetChannel(name string) (*amqp.Channel, error)
  • func (r *RabbitMQClient) GetConnection() *amqp.Connection
  • func (r *RabbitMQClient) IsConnected() bool
  • func (r *RabbitMQClient) Publish(exchange, routingKey string, body []byte) error
• type RabbitMQServiceProvider
  • func (r *RabbitMQServiceProvider) Boot(ctx context.Context) error
  • func (r *RabbitMQServiceProvider) Conf() map[string]string
  • func (r *RabbitMQServiceProvider) Description() string
  • func (r *RabbitMQServiceProvider) HealthCheck(ctx context.Context) error
  • func (r *RabbitMQServiceProvider) Register(ctx context.Context) error
  • func (r *RabbitMQServiceProvider) SetApplication(app foundation.Application)
  • func (r *RabbitMQServiceProvider) Shutdown(ctx context.Context) error
• type RetryConfig
• type TLSConfig

Types

type ConnectionConfig

type ConnectionConfig struct {
	Host                 string `json:"host"`
	Port                 int    `json:"port"`
	Username             string `json:"username"`
	Password             string `json:"password"`
	VHost                string `json:"vhost"`
	Timeout              int    `json:"timeout"`
	Heartbeat            int    `json:"heartbeat"`
	TLS                  bool   `json:"tls"`
	MaxReconnectAttempts int    `json:"max-reconnect-attempts"`
	ReconnectInterval    int    `json:"reconnect-interval"`
}

ConnectionConfig 连接配置

type ConsumerConfig

type ConsumerConfig struct {
	Tag            string `json:"tag"`
	AutoAck        bool   `json:"auto-ack"`
	Exclusive      bool   `json:"exclusive"`
	NoWait         bool   `json:"no-wait"`
	PrefetchCount  int    `json:"prefetch-count"`
	PrefetchSize   int    `json:"prefetch-size"`
	GlobalPrefetch bool   `json:"global-prefetch"`
}

ConsumerConfig 消费者配置

type ExchangeConfig

type ExchangeConfig struct {
	DefaultName string `json:"default-name"`
	DefaultType string `json:"default-type"`
	Durable     bool   `json:"durable"`
	AutoDelete  bool   `json:"auto-delete"`
	Internal    bool   `json:"internal"`
	NoWait      bool   `json:"no-wait"`
}

ExchangeConfig 交换机配置

type MessageConfig

type MessageConfig struct {
	ContentType     string `json:"content-type"`
	ContentEncoding string `json:"content-encoding"`
	DeliveryMode    uint8  `json:"delivery-mode"`
	Priority        uint8  `json:"priority"`
	Expiration      string `json:"expiration"`
}

MessageConfig 消息配置

type MonitoringConfig

type MonitoringConfig struct {
	Enabled            bool `json:"enabled"`
	Interval           int  `json:"interval"`
	MonitorQueues      bool `json:"monitor-queues"`
	MonitorExchanges   bool `json:"monitor-exchanges"`
	MonitorConnections bool `json:"monitor-connections"`
}

MonitoringConfig 监控配置

type Options

type Options struct {
	Connection ConnectionConfig `json:"connection"`
	TLS        TLSConfig        `json:"tls"`
	Pool       PoolConfig       `json:"pool"`
	Exchange   ExchangeConfig   `json:"exchange"`
	Queue      QueueConfig      `json:"queue"`
	Consumer   ConsumerConfig   `json:"consumer"`
	Producer   ProducerConfig   `json:"producer"`
	Message    MessageConfig    `json:"message"`
	Retry      RetryConfig      `json:"retry"`
	Monitoring MonitoringConfig `json:"monitoring"`
}

Options RabbitMQ 配置选项

type PoolConfig

type PoolConfig struct {
	MaxConnections     int `json:"max-connections"`
	MinIdleConnections int `json:"min-idle-connections"`
	IdleTimeout        int `json:"idle-timeout"`
	MaxLifetime        int `json:"max-lifetime"`
}

PoolConfig 连接池配置

type ProducerConfig

type ProducerConfig struct {
	Mandatory   bool `json:"mandatory"`
	Immediate   bool `json:"immediate"`
	ConfirmMode bool `json:"confirm-mode"`
	Timeout     int  `json:"timeout"`
}

ProducerConfig 生产者配置

type QueueConfig

type QueueConfig struct {
	NamePrefix string                 `json:"name-prefix"`
	Durable    bool                   `json:"durable"`
	AutoDelete bool                   `json:"auto-delete"`
	Exclusive  bool                   `json:"exclusive"`
	NoWait     bool                   `json:"no-wait"`
	Args       map[string]interface{} `json:"args"`
}

QueueConfig 队列配置

type RabbitMQClient

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

RabbitMQClient RabbitMQ 客户端包装器

func NewRabbitMQ

func NewRabbitMQ(opt *Options, l log.Logger) *RabbitMQClient

NewRabbitMQ 初始化 RabbitMQ 客户端

func (*RabbitMQClient) BindQueue

func (r *RabbitMQClient) BindQueue(queueName, routingKey, exchangeName string) error

BindQueue 绑定队列到交换机

func (*RabbitMQClient) Connect

func (r *RabbitMQClient) Connect() error

Connect 连接到 RabbitMQ 服务器

func (*RabbitMQClient) Consume

func (r *RabbitMQClient) Consume(queueName string, handler func(amqp.Delivery)) error

Consume 消费消息

func (*RabbitMQClient) DeclareExchange

func (r *RabbitMQClient) DeclareExchange(name, kind string) error

DeclareExchange 声明交换机

func (*RabbitMQClient) DeclareQueue

func (r *RabbitMQClient) DeclareQueue(name string) (amqp.Queue, error)

DeclareQueue 声明队列

func (*RabbitMQClient) Disconnect

func (r *RabbitMQClient) Disconnect() error

Disconnect 断开连接

func (*RabbitMQClient) GetChannel

func (r *RabbitMQClient) GetChannel(name string) (*amqp.Channel, error)

GetChannel 获取或创建通道

func (*RabbitMQClient) GetConnection

func (r *RabbitMQClient) GetConnection() *amqp.Connection

GetConnection 获取原始连接

func (*RabbitMQClient) IsConnected

func (r *RabbitMQClient) IsConnected() bool

IsConnected 检查连接状态

func (*RabbitMQClient) Publish

func (r *RabbitMQClient) Publish(exchange, routingKey string, body []byte) error

Publish 发布消息

type RabbitMQServiceProvider

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

func (*RabbitMQServiceProvider) Boot

func (r *RabbitMQServiceProvider) Boot(ctx context.Context) error

func (*RabbitMQServiceProvider) Conf

func (r *RabbitMQServiceProvider) Conf() map[string]string

func (*RabbitMQServiceProvider) Description

func (r *RabbitMQServiceProvider) Description() string

func (*RabbitMQServiceProvider) HealthCheck

func (r *RabbitMQServiceProvider) HealthCheck(ctx context.Context) error

func (*RabbitMQServiceProvider) Register

func (r *RabbitMQServiceProvider) Register(ctx context.Context) error

func (*RabbitMQServiceProvider) SetApplication

func (r *RabbitMQServiceProvider) SetApplication(app foundation.Application)

func (*RabbitMQServiceProvider) Shutdown

func (r *RabbitMQServiceProvider) Shutdown(ctx context.Context) error

type RetryConfig

type RetryConfig struct {
	Enabled           bool    `json:"enabled"`
	MaxAttempts       int     `json:"max-attempts"`
	Interval          int     `json:"interval"`
	BackoffMultiplier float64 `json:"backoff-multiplier"`
	MaxInterval       int     `json:"max-interval"`
}

RetryConfig 重试配置

type TLSConfig

type TLSConfig struct {
	Enabled            bool   `json:"enabled"`
	CertFile           string `json:"cert-file"`
	KeyFile            string `json:"key-file"`
	CAFile             string `json:"ca-file"`
	InsecureSkipVerify bool   `json:"insecure-skip-verify"`
}

TLSConfig TLS/SSL 配置