IndustryKit MQTT
MQTT source 连接 Broker 并等待消息，不主动轮询设备。采集频率由发布端决定。 支持 MQTT 3.1.1、TCP/TLS、QoS 0/1、JSON 和 text payload。
配置层级¶
一个 MQTT 采集实例由三层配置组成：
[[sources]]定义采集实例、默认 measurement 和公共标签。[sources.mqtt]定义 Broker、会话、认证和 TLS。- 每个
[[sources.mqtt.topics]]定义一条 topic 到 metric 或 logging 的映射，可重复配置多个 topic。
topic 下的 fields、tags 和 topic_tags 是映射表，可写成示例中的内联表，也可写成
[sources.mqtt.topics.fields] 等子表；子表属于它前面的那一个 [[sources.mqtt.topics]]。
type = "mqtt" 时只能配置 [sources.mqtt] 这一种协议子表，并且至少配置一个 topic。
下面的完整片段需追加到 IndustryKit 主配置中。
配置¶
[[sources]]
id = "mqtt-gateway"
type = "mqtt"
measurement = "mqtt_metrics"
tags = { gateway = "gw-01" }
[sources.mqtt]
brokers = ["tcp://192.0.2.30:1883"]
client_id = "industrykit-plant-a"
connect_timeout = "10s"
keep_alive = "30s"
[[sources.mqtt.topics]]
filter = "plant/+/telemetry"
qos = 1
category = "metric"
payload_format = "json"
fields = { temperature = "/values/temperature", speed = "/values/speed" }
timestamp_pointer = "/ts"
timestamp_unit = "ms"
topic_tags = { area = 1 }
[[sources.mqtt.topics]]
filter = "plant/+/events"
qos = 1
category = "logging"
payload_format = "json"
message_pointer = "/message"
status_pointer = "/status"
topic_tags = { area = 1 }
使用 TLS 和用户名认证时，将 [sources.mqtt] 改为类似配置：
[sources.mqtt]
brokers = ["ssl://192.0.2.31:8883"]
client_id = "industrykit-plant-a"
username = "<MQTT_USERNAME>"
password = "<MQTT_PASSWORD>"
ca_file = "pki/mqtt-ca.pem"
cert_file = "pki/mqtt-client.pem"
key_file = "pki/mqtt-client-key.pem"
connect_timeout = "10s"
keep_alive = "30s"
参数¶
[[sources]]¶
| 字段 | 默认值 | 说明 |
|---|---|---|
id |
无 | 必填且在当前配置中唯一；作为 source_id 标签上报。 |
type |
无 | 必须为 mqtt。 |
measurement |
industrial |
topic 未覆盖时使用的指标集或日志来源名称。 |
interval |
10s |
MQTT 为被动接收，该值不控制消息到达或上报频率。 |
tags |
{} |
当前 source 的静态标签，会与全局和 topic 标签合并。 |
[sources.mqtt]¶
| 字段 | 默认值 | 说明 |
|---|---|---|
brokers |
无 | 必填的非空数组；每项为 tcp://host:port、ssl://host:port 或 tls://host:port，必须显式填写有效端口，且不能带路径、账号、query 或 fragment。 |
client_id |
<collector.id>-<source.id> |
持久会话 ID；同一 Broker 和传输方式上必须唯一，ssl 与 tls 视为同一种传输。 |
username |
空 | 可选的 Broker 用户名。 |
password |
空 | 可选的 Broker 密码；设置后必须同时设置 username。 |
ca_file |
空 | TLS CA PEM 文件；未配置时使用系统信任库。 |
cert_file |
空 | TLS 客户端证书，必须与 key_file 成对配置。 |
key_file |
空 | TLS 客户端私钥，必须与 cert_file 成对配置。 |
connect_timeout |
10s |
连接及订阅超时，必须大于 0。 |
keep_alive |
30s |
MQTT keep-alive 周期，必须大于 0。 |
证书相关相对路径按主配置文件所在目录解析，且文件必须允许 IndustryKit 服务账号读取。 TLS 最低版本为 1.2，Server 名称取 Broker URL 中的主机名。
[[sources.mqtt.topics]]¶
MQTT 不使用 point，因此没有单独的 field 配置。JSON metric 中 fields 的左侧 key
就是输出字段名，右侧 value 是从 payload 取值的 JSON Pointer；text metric 使用
value_field 指定输出字段名。
| 字段 | 默认值 | 说明 |
|---|---|---|
filter |
无 | 必填且在当前 source 中唯一；+ 必须独占一层，# 必须独占最后一层。 |
qos |
0 |
订阅 QoS：0 或 1。 |
category |
metric |
输出类型：metric 或 logging。 |
payload_format |
json |
payload 类型：json 或 text。 |
measurement |
继承 source | 覆盖当前 topic 的指标集或日志来源名称。 |
fields |
{} |
metric + json 时必填的非空映射：输出字段名 = JSON Pointer。Pointer 遵循 RFC 6901。 |
value_field |
value |
metric + text 时的输出字段名；整个 payload 作为字符串写入该字段。 |
message_pointer |
空 | logging + json 时必填，定位日志 message。 |
status_pointer |
空 | logging + json 时可选，定位日志 status；省略时 status 为 info。 |
timestamp_pointer |
空 | 可选的消息时间 JSON Pointer；省略时使用接收时间。 |
timestamp_unit |
设置 timestamp pointer 时为 ms |
数值时间戳单位：s、ms、us 或 ns；RFC 3339 字符串时间不使用该值。 |
tags |
{} |
当前 topic 的静态标签。不能配置 collector_id、protocol、source_id 或 mqtt_topic。 |
topic_tags |
{} |
tag 名 = topic 零基层级下标；下标必须非负，越界时消息无效。固定标签及 mqtt_topic 不能作为 tag 名。 |
JSON Pointer 取出的 metric 字段、message 和 status 必须是标量；缺少路径、取到
null、对象或数组都会使该条映射失败并丢弃消息。每个 payload 必须且只能包含一个完整的
顶层 JSON 文档，后面不能再跟第二个 JSON 值。
通用安装、运行和验证方法见 IndustryKit。
数据与限制¶
- 每条数据都会增加
mqtt_topictag。 - text metric 将整个 payload 写入
value_field；text logging 将整个 payload 作为message。 - QoS 1 在所有匹配数据写入 WAL 后 ACK，重试边界可能产生重复；非法消息会丢弃并计数。QoS 0 为尽力而为。
- 单个 MQTT 报文上限为 512 KiB。
- 当前只订阅主题，不发布设备控制消息。