rabbitmq

English | 中文

基于 amqp091-go 构建、面向 go-zero 的 RabbitMQ 客户端，内置 Publisher Confirm、可配置消费并发、处理超时、链路追踪、Prometheus 指标、自动重连 与 优雅停机。

特性

• 📤 Sender / Listener 双端 — Sender 向 Exchange 投递，Listener 从 Queue 消费
• ✅ 投递可靠 — Publisher Confirm + mandatory + 持久化投递：Send 返回 nil 即代表 broker 已确认收下
• 🔀 消费并发可配 — 单队列可选「严格顺序」（Concurrency=1）或「并发处理」（Concurrency=N）
• ⏱️ 处理超时 — 单条消息可设 deadline，支持由宿主框架统一注入
• 🔗 链路追踪 — 集成 OpenTelemetry，生产端注入 Span，消费端提取并延续链路
• 📊 可观测性 — 内置 20 项 Prometheus 指标，经 go-zero /metrics 暴露
• 🔄 自动重连 — 连接/通道关闭后指数退避重连，且可被停机信号中断
• 🛡️ 优雅停机 — 先停止取新消息，再等待在途消息处理完，整体受 GracefulTimeout 兜底
• 🧩 统一中间件 — 消费与投递共用一套 Middleware 抽象，业务中间件可追加注入
• 🔐 TLS — 支持单向与双向 TLS
• 🏗️ Admin — 声明 Exchange、Queue 与绑定关系

安装

go get github.com/lerity-yao/czt-contrib/mq/rabbitmq

服务生成

与 goctl 类似，可以用 cztctl（goctl 的定制分支）一条命令生成服务端代码，同时支持自定义代码生成模板。

详见 cztctl。

快速开始

消费者（Listener）

package main

import (
	"context"
	"time"

	"github.com/lerity-yao/czt-contrib/mq/rabbitmq"
	"github.com/zeromicro/go-zero/core/logx"
)

func main() {
	conf := rabbitmq.RabbitListenerConf{
		RabbitConf: rabbitmq.RabbitConf{
			Username: "guest",
			Password: "guest",
			Host:     "127.0.0.1",
			Port:     5672,
		},
		ListenerQueues: []rabbitmq.ConsumerConf{
			{Name: "queue.demo", Concurrency: 5, Timeout: 3000},
		},
		ChannelQos: rabbitmq.ChannelQosConf{PrefetchCount: 10},
	}

	listener := rabbitmq.MustNewListener(conf, rabbitmq.HandlerFunc(
		func(ctx context.Context, meta rabbitmq.Meta, message []byte) error {
			logx.WithContext(ctx).Infof("received from %s: %s", meta.Queue, message)
			return nil
		}))

	defer listener.Stop()
	listener.Start() // 阻塞直到 Stop 被调用
}

生产者（Sender）

sender := rabbitmq.MustNewSender(rabbitmq.RabbitSenderConf{
	RabbitConf: rabbitmq.RabbitConf{
		Username: "guest",
		Password: "guest",
		Host:     "127.0.0.1",
		Port:     5672,
	},
	ContentType: "application/json",
})
defer sender.Close()

if err := sender.Send(ctx, "exchange.demo", "route.demo", []byte(`{"id":1}`)); err != nil {
	// 返回 nil 才代表 broker 已确认收下
	logx.WithContext(ctx).Errorf("send failed: %v", err)
}

配置参数

RabbitConf（基础连接配置）

字段	类型	默认	说明
Username	string	—	账号，必填
Password	string	—	密码，必填
Host	string	—	地址，必填
Port	int	—	端口，必填，取值须在 1~65535
VHost	string	空	命名空间
Heartbeat	int64	30	心跳间隔（秒），用于探测死链触发重连
TLS	TLSConf	关闭	TLS 配置

TLSConf（TLS 配置）

字段	类型	默认	说明
Enable	bool	false	启用后拨号 scheme 由 amqp 切换为 amqps
CACertFile	string	空	CA 证书路径，为空时使用系统根证书池
ClientCertFile	string	空	客户端证书路径，用于双向认证
ClientKeyFile	string	空	客户端私钥路径，与 ClientCertFile 须同时提供或同时为空
InsecureSkipVerify	bool	false	跳过服务端证书校验，仅供自签证书的测试环境使用

RabbitSenderConf（生产端配置）

内嵌 RabbitConf，另有：

字段	类型	默认	说明
ContentType	string	text/plain	投递报文的 MIME 类型
PoolSize	int	10	confirm channel 池大小，投递吞吐上限约为 PoolSize / 网络 RTT
PublishTimeout	int64	5000	一次 Send 的整体超时（毫秒），含借用 channel 的排队、等待确认与全部重试
MaxRetry	int	3	可重试故障的重试次数，0 表示不重试
GracefulTimeout	int64	10000	优雅关闭超时（毫秒）：Close 等待在途投递收到 broker 确认的上限

RabbitListenerConf（消费端配置）

内嵌 RabbitConf，另有：

字段	类型	默认	说明
ListenerQueues	[]ConsumerConf	—	待监听的队列列表，必填且非空；每个队列独占一个 channel
ChannelQos	ChannelQosConf	—	消费通道 QoS，对所有队列生效
GracefulTimeout	int64	10000	优雅等待超时（毫秒）：重连/停止时等待在途消息处理完成的上限

ConsumerConf（队列消费配置）

字段	类型	默认	说明
Name	string	—	队列名称，必填且不得重复
AutoAck	bool	false	true 时 broker 投递即确认，框架不再做任何确认动作
Exclusive	bool	false	队列独占模式
NoWait	bool	false	true 为非阻塞模式，不等待 broker 响应，无法感知消费者是否创建成功
Concurrency	int	1	单队列消费并发度，上限 256
Timeout	int64	0	单条消息处理超时（毫秒），0 表示不限制。传入 WithConsumeTimeout 时本字段被覆盖

Concurrency 的两种模式：

• 等于 1：严格顺序模式，同一队列的消息按 broker 投递顺序串行处理
• 大于 1：并发模式，启动对应数量的 worker 协程并行处理，队列内不再保证顺序

约束：Concurrency <= ChannelQos.PrefetchCount，否则多余的 worker 永远拿不到消息。该约束由 Validate 强制。

ChannelQosConf（通道 QoS 配置）

字段	类型	默认	说明
PrefetchCount	int	5	单队列未确认消息的上限，达到上限后 broker 停止投递；上限 65535
PrefetchSize	int	0	RabbitMQ 未实现，必须为 0，否则 Validate 报错

PrefetchCount 不是并发度。它是流控参数（由消息大小与网络 RTT 决定），并发度由 Concurrency 决定（由下游承载能力决定），两者职责不同。

PrefetchCount 是每队列的上限，不是整个 Listener 的。 QoS 逐 channel 设置，而每个队列独占一个 channel，故 N 个队列时在途消息上限为 PrefetchCount × N。内存占用按 PrefetchCount × 队列数 × 单条消息大小 估算，队列数多时需相应下调。

ExchangeConf / QueueConf（声明配置）

ExchangeConf 字段	类型	默认	说明
ExchangeName	string	—	交换机名称
Type	string	—	direct / fanout / topic / headers
Durable	bool	true	持久化
AutoDelete	bool	false	无绑定时自动删除
Internal	bool	false	内部交换机
NoWait	bool	false	不等待 broker 响应
Queues	[]QueueConf	—	关联队列

QueueConf 字段	类型	默认	说明
Name	string	—	队列名称
Durable	bool	true	持久化
AutoDelete	bool	false	无消费者时自动删除
Exclusive	bool	false	独占
NoWait	bool	false	不等待 broker 响应

配置校验的三条规则

Validate 对每个字段只按一条规则处理，口径明确：

规则	处理方式	适用情形	例子
A	返回 error（fail-fast）	必填项缺失，或取值不可能正确	Username/Password/Host 为空、Port 越界、队列名重复、PrefetchSize != 0
B	回退默认并记录一行日志	存在安全默认值的非法值	Heartbeat <= 0、Concurrency <= 0、PoolSize <= 0
C	返回 error（上限保护）	超出协议或资源上限	PrefetchCount > 65535、Concurrency > 256、Concurrency > PrefetchCount

所有新增字段均有默认值，配置中心里的现有配置无需任何改动即可升级。

超时体系

层级	配置项	默认	生效方式
单条消息处理	ConsumerConf.Timeout	0（不限）	timeoutMiddleware
宿主注入（消费）	WithConsumeTimeout(d)	—	覆盖全部队列的 Timeout
一次投递	RabbitSenderConf.PublishTimeout	5000	timeoutMiddleware
宿主注入（投递）	WithPublishTimeout(d)	—	覆盖 PublishTimeout
优雅停机	RabbitListenerConf.GracefulTimeout	10000	Stop / reconnect 等待上限
优雅关闭（投递）	RabbitSenderConf.GracefulTimeout	10000	Close 等在途投递收确认的上限

取值优先级：调用方 ctx 自带的 deadline > option（宿主显式传入）> conf 字段 > 默认值。传 0 或负值的 option 视为未设置。

关键语义：配置值只在调用方未给出时限时兜底。timeoutMiddleware 先判断传入的 ctx 是否已带 deadline，已带则原样透传。这与直接 context.WithTimeout(ctx, d) 包一层不同：后者取「较早者」，会把调用方显式要求的更长 deadline 悄悄截断，配置就从兜底变成了强制上限。

⚠️ deadline 是协作式的。handler 必须把 ctx 透传给所有下游阻塞调用，否则超时不生效——time.Sleep 或不接受 ctx 的调用照样会占住 worker 直到执行完毕。

消费侧的消息 ctx 由 Listener 自行从 context.Background() 派生（不得用代次 ctx，否则重连会 cancel 在途消息），本身不带 deadline，因此 Timeout 与 WithConsumeTimeout 在消费侧总是生效。而 Send 的 ctx 由调用方传入，若已带 deadline（如 go-zero rest handler 中），则 PublishTimeout 不介入。

被 go-zero 托管时，在 handler 里显式把宿主的 RestConf.Timeout 注入进来：

return rabbitmq.MustNewListener(
	svcCtx.Config.GDemoARabbitmqConf,
	rabbitmq.HandlerFunc(handler),
	rabbitmq.WithConsumeTimeout(time.Duration(svcCtx.Config.Timeout)*time.Millisecond),
)

投递语义

Send 返回 nil 代表 broker 已确认收下，具体由三重保证支撑：

1. Publisher Confirm — 每次投递等待 broker 的 ack/nack，而非「写入 socket 即返回」
2. mandatory=true — routeKey 无匹配队列时消息被退回并报错，而非被 broker 静默丢弃
3. DeliveryMode=Persistent — 避免「durable 队列 + transient 消息」的半持久化陷阱

错误分流

失败时返回的错误可用 errors.Is 判定原因：

哨兵错误	含义	是否重试	建议处理
ErrPublishUnroutable	无队列绑定该 routeKey，消息已被退回	否	配置错误，应告警而非重投
ErrPublishNacked	broker 明确拒绝（磁盘告警、内部错误等）	是	框架内已退避重试
ErrPublishTimeout	等待确认超时，无法判断消息是否已达 broker	否	由业务决定补偿方式
ErrPublishUnknown	确认到达前 channel 已关闭（多为连接中断），投递结果未知；broker 给出的关闭原因（如 404 NOT_FOUND）会拼在错误末尾	否	同上；先看关闭原因，再排查网络与连接稳定性
ErrSenderClosed	Sender 已关闭	否	停机后的投递，属调用时序问题

分流的分界线是投递结果是否确定。ErrPublishNacked 与 ErrPublishUnroutable 均已收到 broker 的完整回应，结果确定；ErrPublishTimeout 与 ErrPublishUnknown 都没等到确认，消息可能已落地也可能没有，框架对二者一律不重试：重投会在调用方无感的情况下造成重复投递，能否容忍只有业务知道。

switch {
case errors.Is(err, rabbitmq.ErrPublishUnroutable):
	// routeKey 或绑定关系配置错误
case errors.Is(err, rabbitmq.ErrPublishTimeout), errors.Is(err, rabbitmq.ErrPublishUnknown):
	// 消息可能已投递成功，重投会造成重复
}

重试

可重试故障在 PublishTimeout 预算内按 50ms 起步的指数退避重试，最多 MaxRetry 次。预算耗尽即停止，并原样返回最后一次的错误。

只有结果确定的故障才重试：nack（broker 已明确拒收），以及帧尚未发出即失败的情形（借不到 channel、写帧直接被拒）。凡结果未知的一概交回调用方。

⚠️ 重试必然带来重复投递。at-least-once 是本模块的投递保证，消费侧必须幂等——请为消息设计业务唯一键并在消费前查重。

channel 池

amqp091-go 的 Channel 非线程安全，且 NotifyReturn 不携带 deliveryTag，无法在一条 channel 上把多条在途消息的退回归因到正确的那一条。因此并发投递由 channel 池承载，池内每条 channel 同时只处理一条在途消息，并发度由 PoolSize 决定。

收到完整确认（ack、nack 或退回）的 channel 状态干净，直接归还复用；确认超时或连接/channel 故障会留下状态不明的悬空未确认消息，此类 channel 一律销毁重建。

消费语义

Ack 决策

框架层不做重试（沿用「失败就是失败」的定调），业务如需重试请在 handler 中自行实现（延迟队列、记录 DB 等）。

情形	动作	理由
处理成功	Ack	—
业务返回 error	Ack	原样重投只会再失败一次
处理超时	Ack	同上
handler 或中间件 panic	Ack	panic 已就地转为 error：recovery 兼顾业务 handler，逃出链外的中间件 panic 由消费单元兜住
停机拦截（消息未执行）	Reject(requeue=true)	消息尚未被处理，退回队列由其他实例接手
AutoAck=true	不做任何动作	确认权已交给 broker

并发与顺序

Concurrency=N 时，队列的每条 channel 后启动 N 个 worker 协程从同一投递流取消息。顺序模式（N=1）下消息严格按 broker 投递顺序串行处理。

单队列的吞吐上限：顺序模式下为 1 / 单条处理耗时；并发模式下为 N / 单条处理耗时，但受 PrefetchCount 的流控约束。

队列的编排：共用 Listener 还是拆分

一个 Listener 独占一个连接，且有三项配置对其下所有队列统一生效：ChannelQos、GracefulTimeout、WithConsumeTimeout。重连也是连接级的——连接一旦断开，所有队列的消费代次会一起被 cancel 并重建。

当各队列的流控需求与处理耗时在同一量级时，放在一个 Listener 里即可。遇到下列情形则应拆为多个 Listener：

• 预取需求差异大 —— ChannelQos 无法按队列区分，大消息队列与小消息队列只能共用同一个 PrefetchCount
• 处理耗时差异大 —— GracefulTimeout 与 WithConsumeTimeout 是共用的，一个慢队列会替所有快队列定下停机预算
• 需要故障隔离 —— 单一连接故障会重建全部队列，无关队列的抖动会连带关键队列一起停摆

无论是否拆分，每个队列始终保有自己的 channel、Concurrency 与 Timeout；拆分换来隔离性，代价是每个 Listener 多一条 TCP 连接。

中间件

消费与投递共用同一套抽象——Meta 描述消息坐标（消费侧只填 Queue，投递侧只填 Exchange 与 RouteKey），因此不必维护两组同构的类型与链构造器：

type Handler func(ctx context.Context, meta Meta, body []byte) error
type Middleware func(next Handler) Handler

func Chain(middlewares ...Middleware) Middleware

Chain 中靠前的中间件位于外层，最先看到请求、最后看到结果。

内置链的顺序：

• Listener：metrics → logging → consumerTrace → timeout → 业务中间件 → recovery → 业务 handler
• Sender：metrics → logging → producerTrace → timeout → 业务中间件 → recovery → 投递（含重试）

recovery 位于最内层：若置于最外层，panic 会穿透 metrics 与 logging，导致 panic 的消息不计入 consume_total。

追加业务中间件

func auditMiddleware(next rabbitmq.Handler) rabbitmq.Handler {
	return func(ctx context.Context, meta rabbitmq.Meta, body []byte) error {
		start := time.Now()
		err := next(ctx, meta, body)
		logx.WithContext(ctx).Infow("audit",
			logx.Field("queue", meta.Queue),
			logx.Field("cost", time.Since(start).String()),
			logx.Field("err", err))
		return err
	}
}

listener := rabbitmq.MustNewListener(conf, handler,
	rabbitmq.WithConsumeMiddlewares(auditMiddleware))

sender := rabbitmq.MustNewSender(senderConf,
	rabbitmq.WithPublishMiddlewares(auditMiddleware))

业务中间件追加在内置链之后（更贴近业务 handler），因此受处理超时约束。投递侧的中间件位于重试编排之外，因此一次 Send 只执行一次，不会随重试重复触发。

只提供追加、不提供替换：recovery 与 metrics 属框架级保证，允许替换会让调用方在不知情的情况下关掉 panic 兜底与指标采集。

链路追踪

链路上下文走 AMQP 的 header table，遵循 OpenTelemetry 的消息语义约定。生产端将其注入 Meta.Headers，最终落到 amqp.Publishing.Headers；消费端从 amqp.Delivery.Headers 提取并延续链路，业务无需感知。

消息体全程不被触碰，由此带来三点：

• 消息体不必是 JSON。 Protobuf、纯文本、一个裸 ID 都可以，外面不会再套一层。
• 他方投递的消息也能消费。 Java 服务投递的、或从 RabbitMQ 管理界面手工投递的消息不带链路 header，这只意味着没有上游链路可续，消息照样送达 handler。
• 链路可跨语言打通。 任何语言的生产者，只要其 OpenTelemetry SDK 把 traceparent 写入 AMQP header，链路就能在此延续，反向亦然。

传给 handler 的 message 与生产端投递的字节完全一致，ctx 已携带链路信息，直接 logx.WithContext(ctx) 即可串联日志与 trace。

监控指标

经 go-zero /metrics 暴露。

Sender 端

指标	类型	label	用途
rabbitmq_sender_send_total	counter	exchange, route_key, status	投递总数，按结果分类
rabbitmq_sender_send_duration_ms	histogram	exchange, route_key	一次 Send 的端到端耗时
rabbitmq_sender_send_size_bytes	histogram	exchange, route_key	消息体大小
rabbitmq_sender_confirm_duration_ms	histogram	exchange, route_key	等待 broker 确认的耗时
rabbitmq_sender_pool_wait_duration_ms	histogram	endpoint, vhost	借用 channel 的排队耗时
rabbitmq_sender_pool_in_use	gauge	endpoint, vhost	池占用，判断 PoolSize 是否偏小
rabbitmq_sender_retry_total	counter	exchange, route_key	重试次数
rabbitmq_sender_reconnect_total	counter	endpoint, vhost	重连次数
rabbitmq_sender_disconnect_total	counter	endpoint, vhost	断链次数

status 取值：success / fail / timeout / unroutable / nacked / unknown。

Listener 端

指标	类型	label	用途
rabbitmq_listener_consume_total	counter	queue, status	消费总数，按结果分类
rabbitmq_listener_consume_duration_ms	histogram	queue	单条处理耗时
rabbitmq_listener_consume_size_bytes	histogram	queue	消息体大小
rabbitmq_listener_in_flight	gauge	queue	在途消息数，判断 Concurrency 是否偏小
rabbitmq_listener_panic_total	counter	queue	handler panic 次数
rabbitmq_listener_ack_total	counter	queue, type	确认动作，type 为 ack / reject
rabbitmq_listener_reconnect_total	counter	endpoint, vhost	重连次数
rabbitmq_listener_disconnect_total	counter	endpoint, vhost	断链次数

status 取值：ok / fail / timeout / canceled / panic。

endpoint（host:port）与 vhost 区分同一进程内连向不同集群的 Sender / Listener；vhost 未配时归一为 AMQP 默认值 /。

日志前缀

情形	级别	前缀
停机拦截（消息未执行即重入队）	Info	[RABBITMQ_CANCELED]
处理/投递超时	Error	[RABBITMQ_TIMEOUT]
不可路由	Error	[RABBITMQ_UNROUTABLE]
nack	Error	[RABBITMQ_NACKED]
投递重试	Error	[RABBITMQ_RETRY]

自动重连

构造时首次连接只尝试一次：配置错误应在启动时立刻暴露，运行期的连接故障才由重连负责自愈。

监听 Connection 的 NotifyClose，断链后指数退避重连（2s 起步，逐次放大），且退避等待可被停机信号中断。Listener 的一次连接对应一「代」消费协程：重连时先 cancel 旧代并等待其在途消息处理完（上限 GracefulTimeout），再建立新代，据此杜绝新旧两代双重消费。

优雅停机

Listener

Stop() 三段式关停，整体受 GracefulTimeout 兜底，必然返回：

1. 置为停止中状态，新到达的消息一律 Reject(requeue=true) 退回队列
2. cancel 当前代次，等待在途消息处理完毕
3. 关闭 channel 与连接

在途消息的 ctx 不会在停机时被 cancel——那会把「未执行完」误判成业务失败，并制造半途中断的中间态。

Stop 幂等，重复调用无效。被 go-zero ServiceGroup 托管时由框架自动调用。

Sender

构造时通过 proc.AddShutdownListener 自注册停机钩子，进程退出前自动 Close。Close 幂等（显式调用与停机钩子会各调一次），会 cancel 正在退避的重连、清空 channel 池并关闭连接。

Admin

用于在服务启动阶段声明拓扑。Admin 是一次性工具，不做重连也不复用连接——声明动作幂等且只在启动时执行。

⚠️ 用完必须 Close，否则会泄漏一条 AMQP 连接。Close 幂等。

admin, err := rabbitmq.NewAdmin(conf)
if err != nil {
	return err
}
defer admin.Close()

if err = admin.DeclareExchange(rabbitmq.ExchangeConf{
	ExchangeName: "exchange.demo",
	Type:         "direct",
	Durable:      true,
}, nil); err != nil {
	return err
}
if err = admin.DeclareQueue(rabbitmq.QueueConf{Name: "queue.demo", Durable: true}, nil); err != nil {
	return err
}
return admin.Bind("queue.demo", "route.demo", "exchange.demo", false, nil)

go-zero 集成示例

消费端

etc/demoa.yaml：

Name: demoa
Host: 0.0.0.0
Port: 8888
Timeout: 30000                # 注入为单条消息的处理上限

GDemoARabbitmqConf:
  Username: guest
  Password: guest
  Host: 127.0.0.1
  Port: 5672
  GracefulTimeout: 10000
  ChannelQos:
    PrefetchCount: 10
  ListenerQueues:
    - Name: queue.demoa
      Concurrency: 5          # 并发处理；填 1 或省略为严格顺序

internal/config/config.go：

type Config struct {
	rest.RestConf
	GDemoARabbitmqConf rabbitmq.RabbitListenerConf
}

internal/handler/demoA/gdemoahandler.go（cztctl 1.11.0+ 原样生成）：

func GDemoAHandler(svcCtx *svc.ServiceContext) service.Service {
	handler := func(ctx context.Context, meta rabbitmq.Meta, message []byte) error {
		l := demoA.NewGDemoALogic(ctx, meta, svcCtx)
		return l.GDemoA(message)
	}
	return rabbitmq.MustNewListener(
		svcCtx.Config.GDemoARabbitmqConf,
		rabbitmq.HandlerFunc(handler),
		rabbitmq.WithConsumeTimeout(time.Duration(svcCtx.Config.Timeout)*time.Millisecond),
	)
}

meta 注入 Logic 构造函数、而非作为业务方法的入参，因此每个方法里都能取到 l.meta.Queue——这正是一个 handler 服务多个队列的前提。DSL 声明了消息体类型时，生成的 handler 会先反序列化，业务方法拿到的是 *types.XxxReq 而非 []byte。

main.go：

group := service.NewServiceGroup()
defer group.Stop()

group.Add(demoA.GDemoAHandler(ctx))
group.Start()

推荐用 HandlerFunc 闭包的形式：每次消费时新建 Logic 实例，从而拿到本次消费的 ctx（携带链路信息与 deadline），而非复用构造期的 ctx。

完整可运行示例见 example。

生产端

type ServiceContext struct {
	Config     config.Config
	DemoSender rabbitmq.Sender
}

func NewServiceContext(c config.Config) *ServiceContext {
	return &ServiceContext{
		Config:     c,
		DemoSender: rabbitmq.MustNewSender(c.DemoSenderConf),
	}
}

func (l *SendLogic) Send(req *types.SendReq) error {
	body, err := json.Marshal(req)
	if err != nil {
		return err
	}
	return l.svcCtx.DemoSender.Send(l.ctx, "exchange.demo", "route.demo", body)
}

从 v0.1.x 升级

配置无需任何改动，但有以下破坏性变更需要确认。完整清单见 CHANGELOG。

变更	影响
Interceptor / SenderInterceptor / SenderFunc / SenderChain 删除，改为 Meta / Handler / Middleware / Chain	若曾自定义拦截器需改写
Send 改为阻塞等待 confirm，错误细分为五类哨兵	延迟上升、吞吐下降，由 PoolSize 补偿；调用方可按错误类型分流
投递默认 Persistent + mandatory	routeKey 打错会返回错误而非静默成功
ConsumerConf.NoLocal 删除	无影响（RabbitMQ 从未实现该标志，多余的 yaml 键被忽略）
ChannelQosConf.Global 删除	无影响（该标志只在同一 channel 内区分额度归属，而本库每个 channel 上恰好一个 consumer，两种取值行为完全相同；Qos 现固定传 false）
RabbitListenerConf.ContentType 删除	无影响（原为死字段）

建议改动：为存量 handler 补上宿主超时注入，即在 MustNewListener 调用处手工添加 rabbitmq.WithConsumeTimeout(...) 一行（imports 补 "time"）。cztctl 的 handler 模板由远程仓库提供，本模块不介入；且 handler 文件在生成时已存在则跳过，存量服务无论如何都需手工补。

⚠️ rest.RestConf.Timeout 默认仅 3000ms。一旦注入，消息处理就会被隐式限在 3 秒内，消费场景通常需要在 yaml 中显式放大 Timeout。

测试

# 单元测试，不依赖 broker
go test -race .

# 集成测试，需要一个真实的 RabbitMQ
docker run -d --rm -p 5672:5672 rabbitmq:3.13-management
go test -tags=integration -race -timeout=300s .

集成测试的连接参数可由环境变量覆盖：RABBITMQ_HOST / RABBITMQ_PORT / RABBITMQ_USER / RABBITMQ_PASS。

更新日志

见 changelog-cn.md。