rabbitmqx

Documentation

Index

• Constants
• type BindingOptions
• type ConsumerOption
  • func WithConsumerAutoAck(autoAck bool) ConsumerOption
  • func WithConsumerName(name string) ConsumerOption
  • func WithConsumerQueue(queue string) ConsumerOption
• type ConsumerOptions
• type ExchangeOptions
• type ExchangeType
• type Option
• type Options
• type PublisherOption
  • func WithPublisherExchange(exchange string) PublisherOption
  • func WithPublisherMandatory(mandatory bool) PublisherOption
  • func WithPublisherRoutingKey(routingKey string) PublisherOption
• type PublisherOptions
• type QueueOptions
• type RabbitmqConn
  • func NewRabbitmqConn(url string, config *amqp.Config) (*RabbitmqConn, error)
  • func (r *RabbitmqConn) Close()
  • func (r *RabbitmqConn) Consume(opts ConsumerOptions) (<-chan amqp.Delivery, error)
  • func (r *RabbitmqConn) Declare(q *QueueOptions, e *ExchangeOptions, b *BindingOptions) error
  • func (r *RabbitmqConn) Publish(msg amqp.Publishing, opts PublisherOptions) error
• type RabbitmqConsumer
  • func NewRabbitmqConsumer(conn *RabbitmqConn, opts ...ConsumerOption) *RabbitmqConsumer
  • func (r *RabbitmqConsumer) SubscribeMessage(handler func(ctx context.Context, message []byte) error)
• type RabbitmqProducer
  • func NewRabbitmqProducer(conn *RabbitmqConn, opts ...PublisherOption) *RabbitmqProducer
  • func (r *RabbitmqProducer) PublishMessage(ctx context.Context, msg []byte) error
• type Table

Constants

const (
	//Publish模式（发布/订阅模式，消息被路由投递给多个队列，一个消息被多个消费者获取）
	//发布者：交换机
	//订阅者：交换机、队列
	ExchangeTypeFanout = "fanout"

	//Direct模式（路由模式，消息被路由投递给符合路由规则的队列，一个消息被一个消费者获取）
	//发布者：交换机、key
	//订阅者：交换机、队列、key
	ExchangeTypeDirect = "direct"

	//Topic模式（主题模式，消息被路由投递给符合通配符的队列，一个消息被一个消费者获取）
	//发布者：交换机、key
	//订阅者：交换机、队列、key
	ExchangeTypeTopic = "topic"
)

Types

type BindingOptions

type BindingOptions struct {
	RoutingKey string // fanout模式不需要routing key
	NoWait     bool
	Args       Table
}

type ConsumerOption

type ConsumerOption func(*ConsumerOptions)

func WithConsumerAutoAck

func WithConsumerAutoAck(autoAck bool) ConsumerOption

func WithConsumerName

func WithConsumerName(name string) ConsumerOption

func WithConsumerQueue

func WithConsumerQueue(queue string) ConsumerOption

type ConsumerOptions

type ConsumerOptions struct {
	Queue string

	Name      string
	AutoAck   bool
	Exclusive bool
	NoWait    bool
	NoLocal   bool
	Args      Table
}

type ExchangeOptions

type ExchangeOptions struct {
	Name       string
	Kind       string // possible values: empty string for default exchange or direct, topic, fanout
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Passive    bool // if false, a missing exchange will be created on the server
	Args       Table
}

type ExchangeType

type ExchangeType string

ExchangeType 交换机类型

type Option

type Option func(*Options)

Option 是一个函数类型，用来设置配置参数

type Options

type Options struct {
	// 交换机
	ExchangeOptions *ExchangeOptions
	// 队列
	QueueOptions *QueueOptions
	// 绑定
	BindingOptions *BindingOptions
}

type PublisherOption

type PublisherOption func(*PublisherOptions)

func WithPublisherExchange

func WithPublisherExchange(exchange string) PublisherOption

func WithPublisherMandatory

func WithPublisherMandatory(mandatory bool) PublisherOption

func WithPublisherRoutingKey

func WithPublisherRoutingKey(routingKey string) PublisherOption

type PublisherOptions

type PublisherOptions struct {
	Exchange   string
	RoutingKey string

	Mandatory bool // 确保消息成功路由到一个队列，否则返回给生产者。
	Immediate bool // 确保消息立即投递给消费者，如果队列中没有消费者，消息将被退回给生产者。 （已废弃）
}

type QueueOptions

type QueueOptions struct {
	Name       string //不填是随机队列名称
	Durable    bool
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
	Args       Table
}

type RabbitmqConn

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

func NewRabbitmqConn

func NewRabbitmqConn(url string, config *amqp.Config) (*RabbitmqConn, error)

func (*RabbitmqConn) Close

func (r *RabbitmqConn) Close()

func (*RabbitmqConn) Consume

func (r *RabbitmqConn) Consume(opts ConsumerOptions) (<-chan amqp.Delivery, error)

func (*RabbitmqConn) Declare

func (r *RabbitmqConn) Declare(q *QueueOptions, e *ExchangeOptions, b *BindingOptions) error

func (*RabbitmqConn) Publish

func (r *RabbitmqConn) Publish(msg amqp.Publishing, opts PublisherOptions) error

type RabbitmqConsumer

type RabbitmqConsumer struct {
	glog.Logger
	// contains filtered or unexported fields
}

func NewRabbitmqConsumer

func NewRabbitmqConsumer(conn *RabbitmqConn, opts ...ConsumerOption) *RabbitmqConsumer

func (*RabbitmqConsumer) SubscribeMessage

func (r *RabbitmqConsumer) SubscribeMessage(handler func(ctx context.Context, message []byte) error)

type RabbitmqProducer

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

func NewRabbitmqProducer

func NewRabbitmqProducer(conn *RabbitmqConn, opts ...PublisherOption) *RabbitmqProducer

func (*RabbitmqProducer) PublishMessage

func (r *RabbitmqProducer) PublishMessage(ctx context.Context, msg []byte) error

type Table

type Table map[string]interface{}

Table stores user supplied fields of the following types:

bool
byte
float32
float64
int
int16
int32
int64
nil
string
time.Time
amqp.Decimal
amqp.Table
[]byte
[]interface{} - containing above types

Functions taking a table will immediately fail when the table contains a value of an unsupported type.

The caller must be specific in which precision of integer it wishes to encode.

Use a type assertion when reading values from a table for type conversion.

RabbitMQ expects int32 for integer values.