RocketMQ (Prometheus)¶

Apache RocketMQ 是一款分布式消息与流处理平台，可用于业务解耦、流量削峰和事件驱动等场景。

版本支持范围参见 RocketMQ 官方指标文档。使用独立 Exporter 的环境请参考 RocketMQ 集成。

配置¶

前置条件¶

已有运行中的 RocketMQ Broker，版本为 5.1.0 或以上。
能够修改 Broker 实际加载的配置文件并重启服务。集群部署需对每个计划监控的 Broker 开启指标端点。
DataKit 能够访问各 Broker 的指标端口，本文使用默认端口 5557。
已有用于验证的 Topic、消费组和生产消费流量。无业务流量时，部分指标可能为零或没有对应序列。

开启 Broker 内置 Prometheus 指标¶

编辑 Broker 启动参数 -c 实际指向的配置文件，通常为 broker.conf，添加以下配置：

metricsExporterType=PROM
metricsPromExporterHost=0.0.0.0
metricsPromExporterPort=5557

保存后，按现有部署方式重启对应 Broker；生产集群按既有维护流程滚动重启。随后在 Broker 所在主机验证指标端点：

curl -fsS http://127.0.0.1:5557/metrics

返回内容中应包含以 rocketmq_ 开头的指标及其类型信息。还需从 DataKit 主机使用实际 Broker 地址验证端点可达性。

metricsPromExporterHost=0.0.0.0 表示监听所有网络接口，应通过安全组或防火墙限制访问来源。Docker 部署时，需将容器内的 5557 端口映射到 DataKit 可访问的主机端口；如果 DataKit 也运行在容器中，其 127.0.0.1 指向 DataKit 容器自身。

配置 Prom 采集器¶

在 DataKit 主机复制 Prom 采集器示例配置：

mkdir -p /usr/local/datakit/conf.d/prom
cp /usr/local/datakit/conf.d/samples/prom.conf.sample \
  /usr/local/datakit/conf.d/prom/rocketmq.conf

编辑 rocketmq.conf，将文件内容替换为以下配置：

[[inputs.prom]]
  urls = ["http://127.0.0.1:5557/metrics"]
  ignore_req_err = false
  source = "rocketmq"
  # 空列表表示采集全部指标类型，包括 Counter、Gauge 和 Histogram
  metric_types = []
  metric_name_filter = ["^rocketmq_"]
  interval = "10s"
  measurement_prefix = ""
  measurement_name = "rocketmq"
  # 保留完整 Prometheus 指标字段名
  keep_exist_metric_name = true

参数	说明
`urls`	Broker 指标端点地址。`127.0.0.1` 仅适用于 DataKit 与 Broker 同机或共享网络命名空间的情况。
`ignore_req_err`	设置为 `false`，保留指标端点请求错误，便于排查采集异常。
`source`	采集器别名，用于在 DataKit 中区分采集配置。
`metric_types`	空列表采集全部指标类型，避免遗漏 Histogram 指标。
`metric_name_filter`	仅采集以 `rocketmq_` 开头的指标。
`interval`	采集间隔，本文设置为 `10s`。
`measurement_prefix`	指标集前缀，本文设置为空字符串。
`measurement_name`	将采集指标统一写入 `rocketmq` 指标集。
`keep_exist_metric_name`	设置为 `true`，保留完整 Prometheus 指标字段名。

更多参数说明参见 Prom 采集器文档。

集群部署时，将配置中的 urls 替换为各 Broker 的实际指标地址：

urls = [
  "http://<BROKER_1_IP>:5557/metrics",
  "http://<BROKER_2_IP>:5557/metrics"
]

也可在每个 Broker 主机分别部署 DataKit 并采集本机端点。选择一种覆盖方式，避免同一端点重复采集。

保存配置后，重启 DataKit 使配置生效：

datakit service restart

验证采集结果¶

使用 Prom 配置调试命令检查转换后的指标集、字段和标签：

datakit debug --prom-conf /usr/local/datakit/conf.d/prom/rocketmq.conf

调试输出应包含 rocketmq 指标集及以 rocketmq_ 开头的字段。通过以下命令查看采集器运行状态和请求错误：

datakit monitor

指标¶

按照本文配置，指标统一写入 rocketmq 指标集，字段保留完整 Prometheus 名称；实际可采集字段以 Broker /metrics、DataKit 调试和入库结果为准。

指标	描述	类型	单位
`rocketmq_messages_in_total`	Broker 接收的累计消息数，计算每秒速率后可观察生产 TPS。	Counter	条
`rocketmq_messages_out_total`	Broker 投递给消费者的累计消息数，计算每秒速率后可观察消费 TPS。	Counter	条
`rocketmq_throughput_in_total`	Broker 生产写入的累计字节数，计算每秒速率后得到写入吞吐量。	Counter	Byte
`rocketmq_throughput_out_total`	Broker 消费读取的累计字节数，计算每秒速率后得到读取吞吐量。	Counter	Byte
`rocketmq_consumer_ready_messages`	对应 Topic 和消费组尚未被拉取、等待消费的就绪消息数。	Gauge	条
`rocketmq_consumer_inflight_messages`	对应 Topic 和消费组已被拉取、尚未确认完成的消息数。	Gauge	条
`rocketmq_consumer_lag_latency_milliseconds`	当前时间与最早未确认消息的可消费时间之差，反映消费完成的及时性。	Gauge	毫秒
`rocketmq_storage_flush_behind_bytes`	Broker 尚未刷入磁盘的数据量。	Gauge	Byte
`rocketmq_message_size`	发送消息大小的分布，可基于其直方图桶计算 P95。	Histogram	Byte
`rocketmq_rpc_latency_milliseconds`	Broker RPC 调用延迟分布，可按请求类型观察发送、拉取请求的 P95。	Histogram	毫秒

RocketMQ 官方逻辑指标 rocketmq_consumer_lag_latency 和 rocketmq_rpc_latency 在本文示例中带有 _milliseconds 单位后缀。Histogram 指标会展开为 _bucket、_sum 和 _count，例如 rocketmq_rpc_latency_milliseconds_bucket。创建视图和监控器时，应先核对实际字段及单位。

Counter 为累计值，不能直接作为 TPS 或每秒吞吐量展示；计算每秒速率后，消息数单位为条/秒，字节数单位为 Byte/s。Gauge 表示当前状态，就绪消息数、处理中消息数和消费滞后应直接观察其数值与趋势。消费滞后表示最早未确认消息的等待时间，不是已完成消息的平均处理耗时。

常用标签¶

标签	描述
`cluster`	RocketMQ 集群名称。
`node_id`	Broker 节点标识。
`topic`	Topic 名称，用于区分主题维度的生产消费情况。
`consumer_group`	消费组标识，用于定位积压和消费滞后。

不同指标携带的标签不同，以实际采集结果为准；Broker 存储指标通常按 cluster、node_id 查看，消费组指标可按 cluster、topic、consumer_group 查看。