rabbitmq

package
v1.37.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 16, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Index

Constants

View Source
const (
	InitialTimeout      = 5                // 默认失败重新间隔/s
	DefaultExchangeName = "default.change" // 默认交换机
	DefaultExchangeType = "direct"         // 默认交换类型
)

Variables

View Source
var (
	// ErrClosed 表示连接已关闭。
	ErrClosed = errors.New("rabbitmq: connection is closed")
	// ErrNotConnected 表示连接尚未建立或已断开。
	ErrNotConnected = errors.New("rabbitmq: not connected")
	// ErrConnectTimeout 表示重连超过最大尝试次数。
	ErrConnectTimeout = errors.New("rabbitmq: reconnect attempts exhausted")
)

连接状态机错误。

Functions

func MqStart

func MqStart() (err error)

MqStart 创建演示 RabbitMQ 客户端并开始消费固定测试队列。 该函数依赖本地 RabbitMQ 服务,仅用于显式运行的集成演示。

func ReceiveMsg

func ReceiveMsg(queueExchange QueueExchange, receiver Receiver, otherParams ...int) (err error)

ReceiveMsg 接收消息,连接失败后进行重试

func Recv2

func Recv2(mq RabbitMQ, receiver Receiver, taskQuit chan<- struct{})

Recv2 当前监听队列关闭后通知进行重试机制

func Send

func Send(queueExchange QueueExchange, msg interface{}) (err error)

Send 发送消息

func SetGlobal added in v1.36.0

func SetGlobal(connection *Connection)

SetGlobal 设置全局连接(Dial 成功后自动调用;测试可手动设置)。

Types

type Connection

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

Connection 是带状态机的 RabbitMQ 连接。 状态流转:disconnected → connecting → connected;断线后自动按退避重连; Close 后进入 closed,禁止再连接。

func Dial

func Dial(dns string, options ...Option) (*Connection, error)

Dial 建立 RabbitMQ 连接并启动自动重连监控。

func Get added in v1.36.0

func Get() *Connection

Get 获取全局 RabbitMQ 连接;未连接时返回 nil。

func (*Connection) Channel

func (c *Connection) Channel() (*amqp.Channel, error)

Channel 创建新的 AMQP 信道;连接不可用时返回错误。

func (*Connection) Close

func (c *Connection) Close() error

Close 关闭连接并停止自动重连(幂等)。

func (*Connection) IsClosed

func (c *Connection) IsClosed() bool

IsClosed 判断连接是否已关闭。

func (*Connection) State

func (c *Connection) State() State

State 返回当前连接状态。

type Consumer

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

Consumer 消费指定队列并处理消息。 断线后自动等待连接重连并恢复消费;Context 取消后优雅停止。

func NewConsumer

func NewConsumer(conn *Connection, queue QueueExchange, receiver Receiver, options ...ConsumerOption) (*Consumer, error)

NewConsumer 创建消费者;参数校验失败返回错误。

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context) error

Start 消费队列并阻塞,直到 Context 取消或连接关闭。 消费循环内部处理断线重连(依赖 Connection 状态机),消息处理带 panic 恢复。

type ConsumerOption

type ConsumerOption func(*Consumer)

ConsumerOption 是 Consumer 的可选配置。

func WithConsumerMaxRetries

func WithConsumerMaxRetries(maxRetries int) ConsumerOption

WithConsumerMaxRetries 配置消息处理失败的重试次数上限(默认 3)。

type MapChild

type MapChild struct {
	Name string `json:"name"`
	Age  int    `json:"age"`
}

MapChild 是演示消息中的嵌套子对象。

type MapTest

type MapTest struct {
	Name  string   `json:"name"`
	Age   int      `json:"age"`
	Child MapChild `json:"child"`
}

MapTest 是 RabbitMQ 演示消费者解析的消息结构。

type Option

type Option func(*Connection)

Option 是 Connection 的可选配置。

func WithReconnect

func WithReconnect(interval time.Duration, maxRetries int) Option

WithReconnect 配置重连退避间隔与最大尝试次数(maxRetries 为 0 表示不限制)。

type Producer

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

Producer 发送消息到指定交换机/队列。 每次发送使用独立信道,发送后立即关闭,避免信道泄漏。

func NewProducer

func NewProducer(conn *Connection, queue QueueExchange) *Producer

NewProducer 创建生产者。

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, body interface{}) error

Publish 发送普通消息;body 会被 JSON 序列化并标记持久化。

func (*Producer) PublishDelay

func (p *Producer) PublishDelay(ctx context.Context, body string, ttl time.Duration) error

PublishDelay 发送延时消息:消息先进入带 TTL 的延时队列,到期后经死信路由回到原队列。

func (*Producer) PublishRetry

func (p *Producer) PublishRetry(ctx context.Context, body string, retryNums int32, oldRoutingKey, oldExchangeName string) error

PublishRetry 发送重试消息:进入带死信配置的重试队列,TTL 到期后回到原队列。 oldRoutingKey / oldExchangeName 是原队列的绑定信息,用于死信回投。

type QueueExchange

type QueueExchange struct {
	QuName string // 队列名称
	RtKey  string // key值
	ExName string // 交换机名称
	ExType string // 交换机类型
	Dns    string //链接地址
}

QueueExchange 定义队列交换机对象,外部可调用

type RabbitMQ

type RabbitMQ struct {
	Channel *amqp.Channel

	QueueName    string // 队列名称
	RoutingKey   string // key名称
	ExchangeName string // 交换机名称
	ExchangeType string // 交换机类型
	// contains filtered or unexported fields
}

RabbitMQ 定义RabbitMQ对象

var StartQueue *RabbitMQ

StartQueue 保存演示入口创建的 RabbitMQ 客户端。

func NewMq

func NewMq(q QueueExchange) RabbitMQ

NewMq 创建一个新的Mq操作对象,用于延迟消息处理

func (*RabbitMQ) CloseMqChannel

func (r *RabbitMQ) CloseMqChannel() (err error)

CloseMqChannel 关闭rabbitMQ信道

func (*RabbitMQ) CloseMqConnect

func (r *RabbitMQ) CloseMqConnect() (err error)

CloseMqConnect 关闭mq链接

func (*RabbitMQ) MqConnect

func (r *RabbitMQ) MqConnect() (err error)

MqConnect 获取连接给 RabbitMQ

func (*RabbitMQ) MqOpenChannel

func (r *RabbitMQ) MqOpenChannel() (err error)

MqOpenChannel 获取信道给 RabbitMQ

func (*RabbitMQ) SendDelayMsg

func (mq *RabbitMQ) SendDelayMsg(body string, ttl int64) (err error)

SendDelayMsg 生产者发送延时消息

func (*RabbitMQ) SendMsg

func (mq *RabbitMQ) SendMsg(body interface{}) (err error)

SendMsg 生产者发送消息

func (*RabbitMQ) SendRetryMsg

func (mq *RabbitMQ) SendRetryMsg(body string, retry_nums int32, args ...string)

SendRetryMsg 发送重试消息

type Receiver

type Receiver interface {
	Consumer([]byte) error
	FailAction(error, []byte) error
}

Receiver 定义接收者接口

type State

type State int

State 是连接状态机状态。

const (
	StateDisconnected State = iota
	StateConnecting
	StateConnected
	StateClosed
)

连接状态。

func (State) String

func (s State) String() string

String 返回状态可读名称。

type TestReceive

type TestReceive struct {
}

TestReceive 是演示消费者,用于验证消息解析和最终失败处理。

func (*TestReceive) Consumer

func (t *TestReceive) Consumer(messageBody []byte) error

Consumer 解析单条演示消息;JSON 无效时返回包装错误并交由重试流程处理。

func (*TestReceive) FailAction

func (t *TestReceive) FailAction(consumeErr error, messageBody []byte) error

FailAction 返回消息超过重试上限后的最终失败错误。 错误只记录消息长度,不输出可能包含敏感字段的完整消息体。

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL