跳至主要內容

RabbitMQ、RocketMQ 与 Kafka 控制台工具手册

高谷深陵大约 12 分钟约 3578 字
RabbitMQ、RocketMQ 与 Kafka 控制台治理知识海报
RabbitMQ、RocketMQ 与 Kafka 控制台治理参考图把 queue、topic、consumer、lag、message browsing、publish、purge、reset offset、ACL 和脱敏串成 MQ 控制台闭环。高清原图

积压报警响起时，先别点 Reset

支付事件的消费延迟突然升高，值班同学打开控制台，看见红色 lag 后准备点 reset offset。这个动作可能让消息重复消费，也可能跳过尚未处理的数据。排障入口应当先回答四个问题：连的是哪个集群，观察的是哪个 queue/topic 与 consumer group，数字来自哪个时间点，当前身份是否具备写入或管理权限。

先从告警取得环境、集群、资源名、consumer group 和相对时间窗口，再申请个人只读身份。消息 body 往往混有手机号、地址、token、支付信息和内部事件，默认只记录 key、offset 或 msgId、partition/queue、trace id 和脱敏后的必要字段。控制台截图、浏览器缓存和导出的消息样本都按敏感数据处理。

RabbitMQ Management、RocketMQ Dashboard、Kafka UI 和 AKHQ 都是带管理能力的应用，不是天然只读的监控屏。Kafka 自带 CLI 是验证 Kafka UI 数字的基线；RabbitMQ HTTP API 与 rabbitmqadmin、RocketMQ mqadmin 也应作为交叉证据。长期指标应进入 Prometheus 等监控系统，不能让控制台的短期聚合数据承担告警与历史容量分析。

把控制台跑在测试环境

RabbitMQ Management 随 RabbitMQ 发行，但需要在目标节点启用：

rabbitmq-plugins enable rabbitmq_management
rabbitmq-diagnostics -s listeners

启用后无需重启节点。常见 HTTP 监听是 15672，HTTPS 常用 15671，实际值以 listeners 输出和配置为准；这个端口不承载 AMQP、STOMP、MQTT 客户端连接。浏览器访问后先核对集群名与节点列表。Management Plugin还说明 UI/API 会产生统计采集开销，实体数量很大时要评估采集间隔与页面查询成本。

RocketMQ Dashboard 当前稳定版为 2.1.0，可用官方镜像在测试网络启动：

docker run -d --name rocketmq-dashboard \
  -p 127.0.0.1:18080:8080 \
  -e "JAVA_OPTS=-Drocketmq.namesrv.addr=namesrv.example.test:9876" \
  apacherocketmq/rocketmq-dashboard:2.1.0

生产还应把标签解析成经过审核的镜像 digest，不能长期跟随 latest。2.1.0 源码构建和运行基线已经是 Java 17，旧版文档中的 JDK 8 示例不能继续作为当前构建依据。rocketmq.namesrv.addr 决定发现哪个集群，Dashboard 还必须能从容器网络访问 NameServer 返回的 Broker 地址。只通 9876 而 Broker 地址不可达，会出现页面能开、Topic 能列出、消息或管理操作失败的半连通状态。

Kafka 的 CLI 随 Kafka 发行包提供。使用与目标集群兼容的 CLI，并准备不入库的 client.properties：

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="console_ro" password="${SECRET_FROM_LOCAL_STORE}";
ssl.truststore.location=/secure/path/kafka.truststore.jks

配置文件里的占位表达式不会被所有 Kafka CLI 自动解析，实际注入要由受控脚本或本地密钥机制完成，生成的临时文件设最小文件权限并及时销毁。当前稳定基线为 Kafka 4.3.1，本地 CLI 使用 Java 17 或更高版本，并以目标集群兼容矩阵为准；具体发行线应查对应的 Java version 页面，不用“本机能启动”替代兼容性检查。

kafka-topics.sh --bootstrap-server kafka-dev.example.test:9092 \
  --command-config client.properties --list
kafka-consumer-groups.sh --bootstrap-server kafka-dev.example.test:9092 \
  --command-config client.properties --group app-order-worker-dev --describe

原 provectuslabs/kafka-ui 项目已由 Kafbat 团队延续，当前 Kafbat UI 稳定版为 1.5.0，镜像入口是 ghcr.io/kafbat/kafka-ui:1.5.0；AKHQ 当前稳定版为 0.27.1，镜像可固定为 tchiotludo/akhq:0.27.1。两者均采用 Apache License 2.0，但开源许可不替代镜像漏洞治理、第三方组件清单和企业支持评估。cluster name、bootstrap servers、SASL、TLS、Schema Registry 和认证配置应进入受控部署系统。bootstrapServers 只负责初始发现，客户端随后会连接 broker 广播地址，因此“首页打开但 topic 加载失败”常是 advertised listener、DNS 或容器路由问题。UI 的 OAuth/RBAC 与 Kafka principal/ACL 是两层身份，任何一层配置错误都可能导致过度授权或 403。

配出真正的只读身份

RabbitMQ 的 UI 登录由 user tag 控制，queue、exchange 等资源操作仍由 vhost 的 configure/write/read 正则控制。一个全局观察但不能发布、消费或改拓扑的账号，应使用 monitoring tag，并让三个权限正则都不匹配资源：

# 用户由受控账号流程创建，以下命令不携带密码
rabbitmqctl set_user_tags console_ro monitoring
rabbitmqctl set_permissions -p / console_ro '^$' '^$' '^$'
rabbitmqctl list_user_permissions console_ro

rabbitmqctl add_user USER PASSWORD 会把密码放进进程参数和 shell history，因此共享主机上不把占位符替换为真实密码直接执行；账号应由密钥感知的自动化、受保护的 HTTP API 请求或外部身份系统创建。权限变更后重新登录复验。administrator tag 也不会自动绕过普通资源权限，但它能管理用户、vhost 和权限，不应给日常观察账号。还要注意 RabbitMQ 的 queue.purge 与 basic.get/basic.consume 都要求队列的 read 权限：一旦为了看消息临时授予 read，这个身份也可能清空 ready 消息。生产消息浏览与 purge 若要分权，必须再通过独立账号、网关/API 白名单或审批代理隔离，RabbitMQ 的三个资源正则本身无法把这两个动作拆开。

RocketMQ Dashboard 本身不能被假定为强认证边界。RocketMQ 5.3.0+ 提供 ACL 2.0，5.3.3 起不再支持 ACL 1.0；旧集群与新集群的凭证和权限配置不能混抄。官方 Security明确提醒 Dashboard 默认不提供强认证，因此至少放在内网/VPC，由 VPN、Ingress/OAuth、IP allowlist 和 RocketMQ ACL 共同保护。accessKey/secretKey 进入密钥系统，Dashboard 配置、容器环境和启动日志都要检查泄露。

Kafka 的最终授权由 broker authorizer 和 ACL 决定。只读观察通常需要目标 Topic 的 Describe，浏览消息还需要 Read，查看 consumer group 需要 Group 的 Describe 或相应权限；不要授予 Write、Create、Delete、Alter、AlterConfigs。若集群设置 allow.everyone.if.no.acl.found=true，没有 ACL 的资源会改变默认拒绝语义，必须纳入安全审查。UI 内再建立 observer 角色隐藏 produce、topic management 和 offset mutation，但不能以隐藏按钮代替 Kafka ACL。

AKHQ 若用于生产消息浏览，优先使用 json_mask_by_default，只显式开放可见字段；Schema Registry 不可用或消息无法结构化时，应宁可显示占位内容，不退回展示原始二进制。Kafka UI 的 data masking 也要用虚构消息做回归，因为反序列化器、schema 演进和自定义 serde 都可能绕开预想字段。

三组正反实验

先为实验创建专用的 app.console.lab Topic/Queue 和 console-lab consumer group，确保没有真实业务消费者。RabbitMQ 正向读取健康与队列列表时，不把密码放进 curl -u 的进程参数：

umask 077
read -rsp 'RabbitMQ password: ' RMQ_PASSWORD; printf '\n'
cat > .rabbit-console.curl <<EOF
user = "console_ro:${RMQ_PASSWORD}"
cacert = "/secure/path/enterprise-ca.pem"
EOF
unset RMQ_PASSWORD
trap 'rm -f .rabbit-console.curl' EXIT

curl --fail --silent --show-error --config .rabbit-console.curl \
  'https://rabbit-dev.example.test:15671/api/overview'
curl --fail --silent --show-error --config .rabbit-console.curl \
  'https://rabbit-dev.example.test:15671/api/queues/%2F'
rm -f .rabbit-console.curl
trap - EXIT

默认 vhost / 在 URI 中编码为 %2F。预期是 200 和正确 cluster_name/队列列表。401 表示认证失败，403 指向 tag 或授权，空列表则检查 vhost 与资源权限。反向实验尝试向实验 exchange publish 或声明 queue，正确结果是 access refused/403；若成功，立刻删除测试消息或拓扑并收紧 write/configure 正则。Management 的 Get messages 会涉及 ack/requeue，观察账号不应靠消费消息来证明“只读”。

RocketMQ 用 mqadmin 交叉验证 Dashboard：

mqadmin clusterList -n namesrv.example.test:9876
mqadmin topicList -n namesrv.example.test:9876
mqadmin consumerProgress -n namesrv.example.test:9876 -g console-lab

预期 cluster、broker、topic 和消费进度与 Dashboard 一致。连接 NameServer 成功但 Broker 请求 timeout，是路由或广播地址证据；CODE: 1 DESC: No route info 指向 Topic、NameServer 或路由尚未注册；ACL 拒绝则检查 Dashboard/mqadmin 是否使用了同一身份。反向实验用只读身份向实验 Topic send，正确结果是权限拒绝。若发送成功，记录 msgId，确认没有业务订阅者后清理实验资源，并修正 Topic/Group 权限。

Kafka 先看 group 状态和 lag：

kafka-consumer-groups.sh --bootstrap-server kafka-dev.example.test:9092 \
  --command-config client.properties --group console-lab --describe

输出的 CURRENT-OFFSET、LOG-END-OFFSET 和 LAG 是逐分区证据。UI 总 lag 对不上时，先核对集群、group、partition、采样时间与 retention，再看 UI 缓存。反向实验用只读 principal 执行 console producer，预期得到 TopicAuthorizationException。出现成功写入则后端 ACL 失败，不能只改 UI 角色。

最后演练 offset reset，但第一步只生成计划：

kafka-consumer-groups.sh --bootstrap-server kafka-dev.example.test:9092 \
  --command-config admin-change.properties \
  --group console-lab --topic app.console.lab \
  --reset-offsets --shift-by -1 --export > reset-plan.csv

预期 CSV 只包含实验 group/topic/partition 和目标 offset。执行前还应保存可回设的原值：

kafka-consumer-groups.sh --bootstrap-server kafka-dev.example.test:9092 \
  --command-config admin-change.properties \
  --group console-lab --topic app.console.lab \
  --reset-offsets --to-current --export > original-offsets.csv

消费者实例必须停用，评审两份 CSV 后才在测试环境把同一条 reset 命令的 --export 改为 --execute。再启动消费者，使用原有唯一 key 核对那一条记录确实被再次处理，而不是重新生产一条新消息冒充“重放”。若 group 仍活跃、目标 offset 超出 retention、分区集合不符或计划波及其他 Topic，停止执行；Kafka 会把越界目标调整到可用 offset 边界，不能把命令成功当成目标值原样生效。需要回设时，在消息仍处于 retention 且消费者停用的条件下使用 --reset-offsets --from-file original-offsets.csv --execute，随后逐分区复查。

RocketMQ 的 resetOffsetByTime 没有与 Kafka --export 等价的通用预演文件。先保存 consumerProgress 的 broker、queue 与 offset 证据，再只对实验 Topic、实验 group 和评审后的时间戳执行；Dashboard 的消费点重置仅适用于集群消费，广播模式不支持，而且其页面路径要求消费者在线。重置后用原 msgId 或业务唯一键证明旧消息再次到达，不能用新发送的消息替代证据。

接入项目与故障证据

仓库应保存控制台契约和无秘密的参数模板，而不是可登录的配置：

cluster: kafka-prod-a
owner: messaging-platform
bootstrap: kafka-prod-a.example.test:9092
auth: personal SASL identity from secret platform
tls: verify broker certificate with enterprise CA
default-role: observer
message-view: masked-by-default
dangerous-actions: produce, purge, delete, alter-config, reset-offset
change-record: required
sample-retention: incident lifetime

值班记录用同一组定位字段串起 UI、CLI 和监控：cluster、resource、group、partition/queue、current offset、log end offset、ready/unacked、采样时间、trace id。不要粘贴完整 body。RabbitMQ 的 ready 与 unacked、RocketMQ 的 broker/queue offset、Kafka 的 committed offset 与 log end offset 模型不同，不能把三个产品都压成一个“积压数”再直接比较。

常见故障也应保留原始证据。TLS 报错保留证书 subject/SAN 和信任链；SASL 失败记录 mechanism 与 principal，不记录 secret；消息乱码记录 content-type、压缩、schema id 和 serde；页面 OOM 或超时记录资源数、消息页大小、JVM/容器限制和请求耗时。这样才能区分身份、网络、序列化、权限与容量问题，而不是一律重启控制台。

清理、回滚与长期治理

实验结束后删除专用 Topic/Queue、consumer group、临时用户和 ACL，先确认资源名只命中实验命名空间。撤销网关临时放行与 VPN 权限，删除 client.properties、offset CSV、消息样本、截图和 shell history 中的凭证，轮换任何可能暴露的 secret。停止并删除本地 Dashboard/UI 容器时，同时移除只属于实验的 volume 和网络；共享部署不能随手删除。

生产变更的回滚要在执行前定义。offset reset 可保存每个 partition/queue 的原 offset，并在消息仍处于 retention 且消费者满足停用或产品在线条件时按原值回设；但已经触发的外部副作用不会随 offset 自动撤销，必须依赖业务幂等、去重和补偿。RabbitMQ purge 只删除队列中 ready 状态的消息，不会召回已经投递且等待确认的消息；RocketMQ 没有一个与 RabbitMQ queue purge 完全等价的日常诊断按钮，删除 Topic 会同时改变路由与元数据；Kafka 的 records delete 会把指定分区的 log start offset 推进到目标位置，也不是“清空页面缓存”。这些动作和删除 Topic/Queue 都没有控制台级撤销，恢复只能依赖仍可用的上游数据、备份或业务补偿，因此不能用来“试一下”。

控制台成本主要来自消息反序列化、全量 Topic/Queue 枚举、频繁统计采集、Schema Registry 调用、浏览大消息和多集群轮询。应限制消息页大小与最大 payload、请求并发、超时、集群数量和容器 CPU/内存，并用监控系统承担长期趋势。控制台自身也要有健康检查、访问日志、版本/digest 清单、漏洞升级节奏和配置回归；RabbitMQ HTTP API 默认不记录每个请求时，需要显式设计审计入口。

架构选型最终看操作模型。RabbitMQ Management 与 broker 集成紧密，适合 vhost/queue 现场诊断，但不宜承担长期监控；RocketMQ Dashboard 覆盖 Topic、consumer、消息和 broker 管理，能力强也意味着网络与 ACL 失守时影响更大；Kafka UI 和 AKHQ 适合多集群可视化与消息浏览，Kafka CLI 则是自动化、审计和交叉验证基线。无论选哪一个，生产默认 observer、危险动作临时授权、后端 ACL 最终裁决、消息默认脱敏、所有变更可追责，才是能长期运行的控制台体系。

跳到正文