不能直接暴露amqp.Publish和amqp.Consume，因会导致业务层耦合连接管理、错误恢复、序列化、重试、死信等非业务逻辑，易引发nil panic、404错误、消息丢失等问题；需封装Broker统一处理连接池、自动声明、编码、超时、流控与ACK策略。
为什么不能直接暴露 amqp.Publish 和 amqp.Consume？
直接把 RabbitMQ 的 amqp.Publish 或 amqp.Consume 暴露给业务层，会导致耦合严重：业务代码得自己处理连接复用、信道错误恢复、JSON 序列化、重试逻辑、死信路由等。模块封装的核心目的不是“多包一层”，而是把重复的、易错的、和业务无关的细节收掉。
实际项目里，常见错误包括：panic: invalid memory address or nil pointer dereference（忘记检查 amqp.Connection 是否已关闭）、Channel error: 404 Not Found（队列未声明就消费）、或 JSON 反序列化失败后消息被丢弃而非转入死信。
- 所有连接和信道必须池化或单例管理，不能每次发布都新建
amqp.Connection - 订阅端必须自动声明 Exchange/Queue/Binding，且支持幂等（多次启动不报错）
- 消息体默认走
json.Marshal，但需允许传入自定义编码器（比如 protobuf） - 发布时若网络中断，应返回明确错误（如
ErrAMQPUndeliverable），而不是静默失败
如何设计模块的初始化和生命周期？
Go 模块不是“一初始化就万事大吉”，RabbitMQ 连接天然脆弱，必须支持断线重连、信道重建、消费者自动恢复。硬编码 amqp.Dial 地址或不设超时，上线后常因 DNS 变更或 TLS 握手卡住导致服务 hang 死。
推荐结构：
立即学习“go语言免费学习笔记（深入）”；
type Broker struct {
conn *amqp.Connection
ch *amqp.Channel
once sync.Once
}
<p>func NewBroker(url string) (*Broker, error) {
b := &Broker{}
if err := b.connect(url); err != nil {
return nil, err
}
return b, nil
}</p><p>func (b <em>Broker) connect(url string) error {
conn, err := amqp.DialConfig(url, amqp.Config{
Heartbeat: 10 </em> time.Second,
Locale: "en_US",
})
if err != nil {
return err
}
ch, err := conn.Channel()
if err != nil {
conn.Close()
return err
}
b.conn = conn
b.ch = ch
return nil
}
-
amqp.DialConfig必须显式设Heartbeat，否则 TCP keepalive 不足以维持连接 - 不要在
NewBroker里直接调conn.Channel()—— 需要sync.Once保证只建一次，且后续失败时可重建 - 模块对外不暴露
*amqp.Connection，只提供Publish()/Subscribe()方法
发布方法怎么统一处理不同交换机类型？
RabbitMQ 的 direct、topic、fanout 行为差异大，但业务层不该感知。比如发日志用 fanout，发订单事件用 topic，硬编码 ch.Publish 的 exchange 类型参数容易出错。
建议按场景封装：
// 发往 topic exchange，自动加 routingKey
err := broker.PublishTopic("order.created", orderEvent)
<p>// 广播到所有绑定队列（fanout）
err := broker.PublishFanout("system.alert", alertMsg)</p><p>// 点对点（direct），精确路由
err := broker.PublishDirect("payment.confirm", paymentID, paymentEvent)
- 每个方法内部自动声明对应 exchange（
if err := ch.ExchangeDeclare(...)），并设置passive: true避免权限冲突 -
PublishTopic必须校验routingKey格式（如不能含空格、长度限制），否则 RabbitMQ 返回406 PRECONDITION_FAILED - 所有发布操作加 context.Context 支持超时和取消：
ch.PublishWithContext(ctx, ...)
订阅端如何避免消息堆积和重复消费？
裸用 amqp.Consume 时，如果 handler 执行慢或 panic，RabbitMQ 会不断重发消息，最终塞爆内存或触发流控。模块必须内置基础流控和 ACK 策略。
关键点：
- 默认使用
amqp.AutoAck: false，且只在 handler 成功返回后才调msg.Ack() - 为每个 consumer 设置
ch.Qos(10, 0, false)—— 限制最多 10 条未确认消息，防止 worker 过载 - handler panic 时，必须
msg.Nack(false, true)（拒绝并重回队首），而非msg.Reject()（可能丢弃） - 支持手动指定
consumerTag，方便在管理界面识别来源（如"svc-payment-processor"）
真正麻烦的是消息去重 —— 模块层面只能做简单幂等（比如检查 msg.Headers["x-dedup-id"]），复杂场景仍需业务层配合生成唯一 ID 并落库判重。
