MQ 部署运维

高谷深陵大约 15 分钟约 4470 字

消息积压意味着生产与处理之间出现了差额，常见原因是消费者停止、某条消息持续失败或下游数据库限流。应用迟迟得不到处理结果时，还可能是消息没有进入正确的队列。处理之前先找出消息所在的位置，再决定恢复连接、增加处理能力，还是重放一段已经保存的消息。

消息位置、消费进度与重放高清原图

把平台接到实际业务链路

三种产品保存的状态不同

平台	数据与路由对象	消费进度	首查位置
RabbitMQ	Exchange、Binding、Queue；消息进入匹配队列	ready、unacked，消费者 Ack 后可移除消息	队列类型、路由、消费者与未确认数
RocketMQ	Topic 分布到 Broker 的 MessageQueue；消息写入存储	消费组在各队列上的位点，另有重试与死信	Topic 路由、Broker 与消费组进度
Kafka	Topic 分成 Partition，记录按分区追加	消费组为各分区提交下一条消费位置	分区 Leader、ISR、起止 offset 和组位点

同一 Kafka 分区可以被多个消费组独立读取；RabbitMQ 普通队列里的竞争消费者共享待处理消息。RocketMQ 的消息模式、重试机制与客户端版本也会改变消费行为。上述区别直接影响“重新消费一遍”如何实施，不能互换管理命令。

初次安装和完整项目接入分别使用 RabbitMQ、RocketMQ、Kafka 主文。已有平台上线检查应从应用配置反向核对真实节点：

生产者 → bootstrap / NameServer / AMQP 入口
        → 实际 Broker 地址与 TLS 身份
        → 指定 vhost、Topic、Queue 和写权限
        → 消息保留、副本和磁盘
消费者 → 组身份、订阅及读取权限
        → 业务处理 → 数据库提交 → Ack / offset 提交

Kafka 的 bootstrap 地址只负责最初发现；元数据中的 advertised listener 仍需从应用网络可达。RocketMQ 查询 NameServer 成功后，还需要访问路由返回的 Broker 或所用客户端要求的 Proxy。RabbitMQ 管理端口可访问，也不能代替 AMQP 端口与 vhost 权限验证。

上线时确定的运行条件

生产、消费、拓扑管理与监控使用不同身份。新应用先在独立命名空间做允许/拒绝验证，再接真实消息；不能为修一条 403 或授权异常给全局管理员权限。TLS 配置必须同时检查服务名和 CA，管理凭据与消息正文不进入公开日志。

为每条重要消息流确定保留时长、允许积压的最老年龄、下游最大处理速率，以及不可重建消息的独立保存方式。发布前核对客户端、Broker 和插件版本与协议兼容；受限环境准备镜像、插件、证书和全部依赖，操作方式见离线部署。

平台存储可以拆为：

控制配置：账号、权限、Topic/Queue、策略、路由
消息数据：日志、队列文件、分区副本
消费状态：组位点、pending / retry / dead-letter
业务结果：数据库流水、幂等键、外部副作用

这些状态不一定保存在同一个地方。仅恢复拓扑后创建出空队列，会得到一个能连接但没有待处理消息的平台。

RabbitMQ：区分待投递、未确认与已移除

在授权节点上运行诊断命令。下面使用 RabbitMQ 4.3 命令口径，/orders 替换为实际 vhost：

rabbitmq-diagnostics alarms
rabbitmqctl list_queues -p /orders \
  name type messages_ready messages_unacknowledged consumers
rabbitmqctl list_connections name user peer_host state channels

ready 持续增加、消费者为零，先查消费者启动与订阅。unacked 增长而 ready 不高，说明消息已送出但尚未完成确认，重点看业务耗时、事务和 prefetch。出现内存或磁盘告警时，生产连接可能被阻塞；先处理资源原因，不无限扩大发送重试。

Ack 应跟随业务提交。消费者提交数据库后、Ack 到达 Broker 前失去连接，消息可能重投，因此需要稳定消息 ID 与数据库唯一约束等去重机制。Consumer acknowledgements 与 publisher confirms说明了两个方向各自确认什么。

用一条消息观察重新入队

以下是 Linux Bash、Docker、curl 与 jq 实验，RabbitMQ 固定 4.3.5-management。只发布到宿主回环地址，新建容器和卷，不复用业务节点。官方镜像入口初始化后由 rabbitmq 身份运行，宿主操作者仍具有 Docker 管理权限。

docker volume create ops3-rabbit-data
docker run -d --name ops3-rabbit --hostname ops3-rabbit \
  --memory 768m -p 127.0.0.1:18672:15672 \
  -e RABBITMQ_DEFAULT_USER=ops_lab \
  -e RABBITMQ_DEFAULT_PASS=LabOnly-Rabbit-83 \
  -v ops3-rabbit-data:/var/lib/rabbitmq rabbitmq:4.3.5-management
docker exec ops3-rabbit rabbitmq-diagnostics -q ping

等待 ping 成功，随后确认管理 API 返回正常 JSON。定义的入口仅用于本机实验；真实共享节点使用 HTTPS 和受限凭据：

rq() {
  curl -q --noproxy '*' --fail-with-body --silent --show-error \
    -u ops_lab:LabOnly-Rabbit-83 -H 'Content-Type: application/json' \
    "http://127.0.0.1:18672$1" "${@:2}"
}
rq /api/overview | jq '.rabbitmq_version'
rq /api/queues/%2F/ops3.verify -X PUT \
  -d '{"durable":true,"auto_delete":false,"arguments":{"x-queue-type":"quorum"}}'
rq /api/exchanges/%2F/amq.default/publish -X POST \
  -d '{"properties":{"delivery_mode":2,"message_id":"ops3-001"},"routing_key":"ops3.verify","payload":"order-102","payload_encoding":"string"}' \
  | jq -e '.routed==true'

%2F 是默认 vhost / 的 URL 编码。发布响应 routed:true 表示管理 API 的这次消息路由成功；AMQP 生产者的 Confirm、Return 和连接超时仍需由真实客户端检查。

用管理 API 取出消息后重新入队：

rq /api/queues/%2F/ops3.verify/get -X POST \
  -d '{"count":1,"ackmode":"ack_requeue_true","encoding":"auto","truncate":100}' \
  | jq -e 'length==1 and .[0].payload=="order-102"'
rq /api/queues/%2F/ops3.verify/get -X POST \
  -d '{"count":1,"ackmode":"ack_requeue_false","encoding":"auto","truncate":100}' \
  | jq -e 'length==1 and .[0].payload=="order-102"'
rq /api/queues/%2F/ops3.verify/get -X POST \
  -d '{"count":1,"ackmode":"ack_requeue_false","encoding":"auto","truncate":100}' \
  | jq -e 'length==0'

第一次读取后消息仍可再次取得；第二次确认并移除，第三次结果为空。API 操作本身会消费消息，只能对这个专用实验队列执行，不能拿生产队列做“只读排查”。字段定义见 RabbitMQ HTTP API。

这种队列消费语义下，Ack 之后不能像 Kafka 一样把消费位点拨回去。需要重新执行业务时，从可靠事件源或业务流水受控重发，并保留原消息 ID 或专门的重放标识。

Definitions 能恢复拓扑，不能补回消息

umask 077
docker exec ops3-rabbit rabbitmqctl \
  export_definitions /tmp/ops3-definitions.json
docker cp ops3-rabbit:/tmp/ops3-definitions.json ./ops3-definitions.json

该文件可能包含用户与权限信息，按敏感配置保存。导入空节点可以恢复队列和路由定义，但上面已经移除的 ops3-001 不会重新出现。

需要冷备消息存储时，按队列类型和集群布局停止相关节点并取得一致备份，保留节点名称、cookie、配置与消息目录。现代 RabbitMQ 磁盘恢复要求相同节点名称；quorum queue 多数派和节点替换也有单独约束，见备份与恢复。不要在运行中复制一个目录后宣称已具备可恢复副本。

实验结束执行 docker stop ops3-rabbit，保留卷。真实集群维护一次只处理满足可用性条件的节点；检查 quorum-critical、队列副本和同步状态后再停机，节点回归并恢复冗余后才处理下一台。

Kafka：保存日志与提交消费位置是两件事

Kafka 标准消费组的提交位点指向下一条要处理的 offset。组位点前进后，分区中的历史记录通常仍按 retention 或 compaction 策略保留；能重放多远由当前可读日志决定。压缩主题也未必保存每次历史变化。

用两个消费组读取同一条记录

下面使用 Apache Kafka 4.3.1 官方镜像的单节点 KRaft 默认实验配置，Linux Bash 下执行。容器不发布宿主端口，客户端通过 docker exec 在内部运行；数据留在这个可丢弃容器中，不作为生产持久化方案。生产集群的独立控制器、认证、日志卷和副本配置见 Kafka 主文。

docker run -d --name ops3-kafka --memory 1g apache/kafka:4.3.1
docker exec ops3-kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --list

服务尚未就绪时先看日志，等列表命令成功再执行创建；不要用格式化存储修复启动等待。

docker exec ops3-kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --topic ops3.verify \
  --partitions 1 --replication-factor 1
printf 'ops3-001:order-102\n' | \
  docker exec -i ops3-kafka /opt/kafka/bin/kafka-console-producer.sh \
    --bootstrap-server localhost:9092 --topic ops3.verify \
    --producer-property acks=all
docker exec ops3-kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic ops3.verify \
  --group ops3-check --from-beginning --max-messages 1 --timeout-ms 10000
docker exec ops3-kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic ops3.verify \
  --group ops3-replay --from-beginning --max-messages 1 --timeout-ms 10000

两个新消费组都应输出 ops3-001:order-102。同一个已有消费组已经提交位点时，--from-beginning 不会强制覆盖已有位置。这里 acks=all 只等待这个单副本实验的同步集合，不能推导出跨节点可靠性。

查看组的位置：

docker exec ops3-kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --describe --group ops3-check
docker exec ops3-kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --describe --topic ops3.verify

比较每个分区的 CURRENT-OFFSET、LOG-END-OFFSET、LAG 和 ISR。位点为空时先确认消费组是否真正提交；不能把“打印过一条消息”直接当成持久提交。Kafka 4.3 的操作与位置管理见 Basic Kafka Operations。

修改位点前，先预览将重放什么

优先用新消费组对隔离下游重放，不改生产组。确实要修改已有组时，先停止该组全部消费者，保存当前每个分区位点，并确认数据仍在保留窗口内。下面仅预览实验组的重置结果：

docker exec ops3-kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --group ops3-check \
  --topic ops3.verify --reset-offsets --to-earliest --dry-run

检查输出是否只包含预期主题与分区、目标 offset 是否正确。正式执行需要将预览方案审批后改为 --execute，执行后再次 describe；不在运行中的消费组上尝试抢改位点。保存的旧位点可以恢复消费位置，但已经重做的业务副作用不会随之撤销。

消费持续落后时，不要用 --to-latest 把数字变成零。这会跳过消息；只有业务明确允许丢弃那段记录并保留缺口处理方式时，才能作此选择。实验结束执行 docker stop ops3-kafka。

RocketMQ：按 Topic 路由和队列位点接管

RocketMQ 5.5 的管理操作在安装目录下执行，使用已授权的管理机与 NameServer 地址；生产鉴权配置按实际认证机制提供，不能把旧 ACL 文件格式直接套到所有 5.x 服务。

export ROCKETMQ_HOME=/opt/rocketmq
export NAMESRV_ADDR='ns-a.example.internal:9876;ns-b.example.internal:9876'
cd "$ROCKETMQ_HOME"
sh bin/mqadmin clusterList -n "$NAMESRV_ADDR"
sh bin/mqadmin topicRoute -n "$NAMESRV_ADDR" -t order-events
sh bin/mqadmin consumerProgress -n "$NAMESRV_ADDR" -g order-worker

Topic 路由返回 Broker 与队列信息，consumerProgress 给出组在队列上的进度。NameServer 通而 Broker 不通时，检查返回地址是否从该管理机和应用可达；组位点不动时，继续查询消费者连接、订阅和失败日志。命令参数见 RocketMQ 管理工具。

由拓扑管理员预先创建全新的空测试 Topic ops3-verify，再做一次带稳定 Key 的收发。使用独立测试组 ops3-reader，命令身份需要对应发送与读取权限：

sh bin/mqadmin sendMessage -n "$NAMESRV_ADDR" \
  -t ops3-verify -k ops3-001 -p order-102

发送结果应为 SEND_OK。记录同一行返回的 Broker Name 和 QID，按这两个值读取对应队列：

read -r -p 'Send result Broker Name: ' TEST_BROKER
read -r -p 'Send result QID: ' TEST_QUEUE
sh bin/mqadmin consumeMessage -n "$NAMESRV_ADDR" \
  -t ops3-verify -g ops3-reader \
  -b "$TEST_BROKER" -i "$TEST_QUEUE" -o 0 -c 1

空队列的第一条消息从 offset 0 读取，结果中应有 Key ops3-001 和正文 order-102。-c 1 不能替代队列定位；在多队列 Topic 上任意读取，可能只看到另一队列的 NO_NEW_MSG。已有数据的 Topic 还须定位目标消息的实际 offset，不能套用这里的 0。管理工具的读取只验证这条测试消息可见，真正业务客户端还要验证成功提交、重试和去重。

普通失败重试、死信消息重投与消费组位点重置分别处理。持续失败的一条消息先保存错误原因、消息标识和订阅配置，修复数据或消费者后用受控重投恢复；直接拉大消费线程数会放大对同一错误的访问。

需要按时间或队列 offset 重置时，先用当前发行版 mqadmin help resetOffsetByTime 核对参数，并保存每队列旧位置。时间到 offset 的映射不是跨队列全局事务位置，重放仍需业务去重；不使用通配主题、不把另一个消费组的进度文件直接覆盖过来。

RocketMQ 的复制方式取决于部署模式。异步主从、同步确认和 Controller 自动切换的承诺不同；接管时核对 Broker 角色、同步状态和客户端实际发送结果，再判断是否允许停下一节点。CommitLog 保留、重试/DLQ、消费组位点和业务流水应一起进入恢复方案。

积压、容量与恢复后的业务检查

用变化速率估算恢复时间

设积压量为 B，稳定生产速率为 P，实际成功处理速率为 C。只有 C 大于 P 时才有条件清空积压，粗略时间为 B / (C - P)。计算使用“业务成功处理”速率，不能把重复投递和失败重试算作有效吞吐。

例如积压 60 万条，生产每秒 1000 条，处理每秒 1500 条，净消化为每秒 500 条，预计约 20 分钟。若数据库只能承受当前写入，再增加消费者会把瓶颈转移到数据库。先限制重放速率、缩短单条处理路径或扩容真正瓶颈。

容量至少覆盖保留期内写入、复制副本、索引/元数据、重试消息及恢复临时空间。不同产品的压缩和清理方式不同，用实测每秒字节增长估算比统一套“每条 1 KB”更可靠。磁盘满时优先停止非必要生产和扩容；不从数据目录手工删除日志段或队列文件。

发送失败、处理失败与数据缺失分开处理

观察	接着检查	修复后的验证
建连超时或 Broker 地址不可达	应用网络到元数据返回地址、端口、DNS 和 TLS	从应用网络重新发送与消费专用消息
认证成功但没有权限	principal、vhost/Topic、消费组和具体动作	允许动作成功，越权动作准确拒绝
Broker 接收受阻	磁盘/内存告警、Leader/ISR/多数派、请求超时	资源恢复、复制稳定，发送不再持续重试
消费在线但积压不降	毒消息、数据库锁、处理耗时、重试和批次	有效处理速率超过生产，最老消息年龄下降
位点已经前进但业务没有结果	是否提前 Ack/提交，业务事务与外部调用记录	由可靠源补齐，重复处理只产生允许的结果
恢复后队列或主题为空	备份是否仅含定义、保留期、复制/归档范围	找到可重放来源，明确缺失区间与业务补偿

对于不可重建消息，保存到独立灾备集群或外部归档。Kafka 跨集群复制需要同时考虑主题配置、权限、组位点转换与延迟，远端 offset 不能直接假定等于源端；具体机制见 Kafka 跨集群镜像。

先恢复依赖，再放大消费

平台恢复后，先确认账户、Topic/Queue、订阅、复制和保留策略；再用独立消息验证生产到消费。随后选择一小段真实待处理数据，核对业务事务、幂等记录、错误率和处理顺序，最后逐步提高并发。

订单取消和支付成功可能分属不同事件。重放时除了消息数，还要检查同一业务键的顺序与最终状态；消息全部读完仍可能得到错误业务结果。DLQ 只有在错误原因已经解决、重投策略明确时才重新入主流，不能把它当成必须立刻清零的指标。

维护升级也遵循这一顺序：先保全配置和数据、验证版本支持的升级路径，逐节点恢复副本与连接，再验证真实生产者和消费者。Kafka 的存储格式/feature、RabbitMQ feature flags、RocketMQ 存储及协议变化都可能限制降级；出现不兼容时暂停下一节点，使用事先验证的独立恢复方案，不能直接换回旧二进制读取新数据目录。

权威资料与规范地址

RabbitMQ

确认机制：https://www.rabbitmq.com/docs/confirms
HTTP API：https://www.rabbitmq.com/docs/http-api-reference
备份与恢复：https://www.rabbitmq.com/docs/backup

Kafka 与 RocketMQ

Kafka 基础运行操作：https://kafka.apache.org/43/operations/basic-kafka-operations/
Kafka 跨集群镜像：https://kafka.apache.org/43/operations/geo-replication-cross-cluster-data-mirroring/
RocketMQ 管理工具：https://rocketmq.apache.org/docs/deploymentOperations/02admintool/