RabbitMQ生产者发不出消息,需检查amqp.Publishing的exchange和routing key是否为空;消费者panic导致消息重复,须关闭autoAck并手动Ack;JSON序列化失败常因字段未导出或tag拼写错误;服务重启后消息堆积,应复用连接/Channel并设置上下文超时。
RabbitMQ 生产者发不出消息?检查 amqp.Publishing 的 exchange 和 routing ke
y 是否为空
很多刚上手的人卡在“消息发了但消费者收不到”,根本原因常是 RabbitMQ 的路由机制没被理解:消息必须先到 Exchange,再根据 routing key 转发到队列。如果生产者调用 ch.PublishWithContext(ctx, "", q.Name, ...) 时第一个参数(exchange)传了空字符串,且没提前声明默认交换机绑定,消息就直接丢弃了——RabbitMQ 不报错,也不提示。
routing key;所以 q.Name 必须和你 Publish 时用的 routing key 完全一致direct 交换机,比如 orders.exchange,再用 ch.ExchangeDeclare(...) + ch.QueueBind(...) 建立绑定autoAck: true 测试生产者——它掩盖了消费者端是否真正连上了队列的问题消费者 panic 后消息重复?必须关掉 autoAck 并手动调用 msg.Ack(false)
Go 消费者若用 ch.Consume(..., autoAck: true),RabbitMQ 一投递就立刻标记消息为已确认,哪怕你的处理逻辑 panic 或崩溃,消息也永远不会重试。线上最常见“订单扣库存两次”问题,根源就是这个开关没关。
autoAck: false,然后在 for msg := range msgs 循环里,仅当业务逻辑完整执行完毕(比如 DB 更新成功、缓存刷新完成)后,才调用 msg.Ack(false)
msg.Nack(false, true) 可用于临时拒收并重回队首;msg.Reject(false) 则直接丢弃(慎用)defer ch.Close() 和连接断开监听(conn.NotifyClose())做兜底,否则 channel 泄露会导致后续无法创建新 consumerJSON 序列化失败却没报错?检查结构体字段是否导出 + json tag 是否拼错
Go 的 json.Marshal 对非导出字段(小写开头)静默忽略,且不返回错误。你看到消息体是 {} 或空字符串,大概率是字段没加 exported 或 json:"order_id" 写成 json:"order_id"(少个引号)或 json:order_id(漏冒号)。
json.Marshal(&order),避免值拷贝时零值字段干扰if len(body) == 0 { log.Fatal("empty JSON body") }
json.Unmarshal(data, &v) 后检查 v.ID 等关键字段是否为零值,早发现字段映射失败服务重启后消息堆积?别只靠重连,要实现连接/Channel 复用 + 上下文超时控制
高频重连(比如每秒 dial 一次)会快速打满 RabbitMQ 的 socket 连接数限制;而每次消费都新建 Channel 又容易触发 AMQP channel limit(默认 65535)。更隐蔽的问题是,没带 context.WithTimeout 的 PublishWithContext 或 Consume 会让 goroutine 卡死在阻塞 IO 上,最终拖垮整个服务。
*amqp.Connection 和 *amqp.Channel 提取为单例或依赖注入对象,复用而非每次新建ch.PublishWithContext, ch.Consume)必须带 context,超时设为 3–5 秒,超时后主动关闭 channel 并重建conn.NotifyClose() 监听连接中断,在回调里触发自动重连 + channel 重建,而不是靠轮询消息队列不是“加个库就能跑”的功能模块,它的可靠性取决于你对 ACK 时机、连接生命周期、序列化边界这三处细节的掌控程度——而这三处,恰恰是日志里几乎不报错、监控里难以定位、压测时才突然爆发的地方。