文档文档

MQTT Consumer 输入插件

此服务插件在支持的 数据格式 之一中,消费 MQTT 代理上指定主题的消息。

引入于: Telegraf v0.10.3 标签: messaging 操作系统支持: all

服务输入

此插件是服务输入。普通插件收集由 interval 设置确定的指标。服务插件启动一个服务来监听并等待指标或事件发生。服务插件与普通插件的两个主要区别是:

  1. 全局或插件特定的 interval 设置可能不适用
  2. --test--test-wait--once 的 CLI 选项可能不会为此插件生成输出

全局配置选项

插件支持其他全局和插件配置设置,用于修改指标、标签和字段,创建别名以及配置插件顺序等任务。更多详情请参阅 CONFIGURATION.md

启动错误行为选项

除了插件特定的和全局的配置设置外,该插件还支持使用 startup_error_behavior 设置来指定出现启动错误时的行为。可用值如下:

  • error:如果出现启动错误,Telegraf 将停止并退出。这是默认行为。
  • ignore:Telegraf 将忽略此插件的启动错误,并禁用它,但会继续处理所有其他插件。
  • retry: Telegraf 会在每次收集或写入周期内尝试启动插件,以防出现启动错误。在启动成功之前,插件将被禁用。
  • probe: Telegraf 将(如果可能)探测插件的功能,并在探测失败时禁用该插件。如果插件不支持探测,Telegraf 将表现得如同设置了 ignore 一样。

Secret-store 支持

此插件支持从 secret-stores 获取 usernamepassword 选项的密钥。有关如何使用它们的更多详细信息,请参阅 secret-store 文档

配置

# Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]
  ## Broker URLs for the MQTT server or cluster.  To connect to multiple
  ## clusters or standalone servers, use a separate plugin instance.
  ##   example: servers = ["tcp://:1883"]
  ##            servers = ["ssl://:1883"]
  ##            servers = ["ws://:1883"]
  servers = ["tcp://127.0.0.1:1883"]

  ## Topics that will be subscribed to.
  topics = [
    "telegraf/host01/cpu",
    "telegraf/+/mem",
    "sensors/#",
  ]

  ## The message topic will be stored in a tag specified by this value.  If set
  ## to the empty string no topic tag will be created.
  # topic_tag = "topic"

  ## QoS policy for messages
  ##   0 = at most once
  ##   1 = at least once
  ##   2 = exactly once
  ##
  ## When using a QoS of 1 or 2, you should enable persistent_session to allow
  ## resuming unacknowledged messages.
  # qos = 0

  ## Connection timeout for initial connection in seconds
  # connection_timeout = "30s"

  ## Interval and ping timeout for keep-alive messages
  ## The sum of those options defines when a connection loss is detected.
  ## Note: The keep-alive interval needs to be greater or equal one second and
  ## fractions of a second are not supported.
  # keepalive = "60s"
  # ping_timeout = "10s"

  ## Max undelivered messages
  ## This plugin uses tracking metrics, which ensure messages are read to
  ## outputs before acknowledging them to the original broker to ensure data
  ## is not lost. This option sets the maximum messages to read from the
  ## broker that have not been written by an output.
  ##
  ## This value needs to be picked with awareness of the agent's
  ## metric_batch_size value as well. Setting max undelivered messages too high
  ## can result in a constant stream of data batches to the output. While
  ## setting it too low may never flush the broker's messages.
  # max_undelivered_messages = 1000

  ## Persistent session disables clearing of the client session on connection.
  ## In order for this option to work you must also set client_id to identify
  ## the client.  To receive messages that arrived while the client is offline,
  ## also set the qos option to 1 or 2 and don't forget to also set the QoS when
  ## publishing. Finally, using a persistent session will use the initial
  ## connection topics and not subscribe to any new topics even after
  ## reconnecting or restarting without a change in client ID.
  # persistent_session = false

  ## If unset, a random client ID will be generated.
  # client_id = ""

  ## Username and password to connect MQTT server.
  # username = "telegraf"
  # password = "metricsmetricsmetricsmetrics"

  ## Optional TLS Config
  # tls_ca = "/etc/telegraf/ca.pem"
  # tls_cert = "/etc/telegraf/cert.pem"
  # tls_key = "/etc/telegraf/key.pem"
  ## Use TLS but skip chain & host verification
  # insecure_skip_verify = false

  ## Client trace messages
  ## When set to true, and debug mode enabled in the agent settings, the MQTT
  ## client's messages are included in telegraf logs. These messages are very
  ## noisey, but essential for debugging issues.
  # client_trace = false

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"

  ## Enable extracting tag values from MQTT topics
  ## _ denotes an ignored entry in the topic path,
  ## # denotes a variable length path element (can only be used once per setting)
  # [[inputs.mqtt_consumer.topic_parsing]]
  #   topic = ""
  #   measurement = ""
  #   tags = ""
  #   fields = ""
  ## Value supported is int, float, unit
  #   [inputs.mqtt_consumer.topic_parsing.types]
  #      key = type

示例输出

mqtt_consumer,host=pop-os,topic=telegraf/host01/cpu value=45i 1653579140440951943
mqtt_consumer,host=pop-os,topic=telegraf/host01/cpu value=100i 1653579153147395661

关于 Topic 解析

整个 MQTT topic 会被存储为一个标签,但这可能过于笼统,在后续使用数据时难以有效利用。此更改允许从 MQTT topic 中提取标签值,从而可以以有意义的方式存储 topic 中提供的信息。下划线 _ 表示 topic 路径中被忽略的条目。请参考以下示例。

Topic 解析示例

[[inputs.mqtt_consumer]]
  ## Broker URLs for the MQTT server or cluster.  To connect to multiple
  ## clusters or standalone servers, use a separate plugin instance.
  ##   example: servers = ["tcp://:1883"]
  ##            servers = ["ssl://:1883"]
  ##            servers = ["ws://:1883"]
  servers = ["tcp://127.0.0.1:1883"]

  ## Topics that will be subscribed to.
  topics = [
    "telegraf/+/cpu/23",
  ]

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "value"
  data_type = "float"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "telegraf/one/cpu/23"
    measurement = "_/_/measurement/_"
    tags = "tag/_/_/_"
    fields = "_/_/_/test"
    [inputs.mqtt_consumer.topic_parsing.types]
      test = "int"

将生成以下指标

cpu,host=pop-os,tag=telegraf,topic=telegraf/one/cpu/23 value=45,test=23i 1637014942460689291

字段枢轴示例

您可以使用 pivot processor 将单值指标旋转为多字段指标。有关更多信息,请查看 pivot processor

在此示例中,这些是 topic

/sensors/CLE/v1/device5/temp
/sensors/CLE/v1/device5/rpm
/sensors/CLE/v1/device5/ph
/sensors/CLE/v1/device5/spin

这些是指标

sensors,site=CLE,version=v1,device_name=device5,field=temp value=390
sensors,site=CLE,version=v1,device_name=device5,field=rpm value=45.0
sensors,site=CLE,version=v1,device_name=device5,field=ph value=1.45

在配置中使用 pivot 将把指标旋转为多字段指标。配置

[[inputs.mqtt_consumer]]
    ....
    topics = "/sensors/#"
    [[inputs.mqtt_consumer.topic_parsing]]
        measurement = "/measurement/_/_/_/_"
        tags = "/_/site/version/device_name/field"
[[processors.pivot]]
    tag_key = "field"
    value_key = "value"

将生成以下指标

sensors,site=CLE,version=v1,device_name=device5 temp=390,rpm=45.0,ph=1.45

Metrics

  • 所有测量结果都将带有传入 topic 的标签,例如 topic=telegraf/host01/cpu

  • 当设置 [[inputs.mqtt_consumer.topic_parsing]] 时的示例

  • 当设置 [[inputs.internal]] 时的示例

    • payload_size (int): 获取从传入消息接收到的累计字节大小
    • messages_received (int): 从 mqtt 接收到的消息数量计数

这将生成以下指标

internal_mqtt_consumer host=pop-os version=1.24.0 messages_received=622i payload_size=37942i 1657282270000000000

此页面是否有帮助?

感谢您的反馈!


InfluxDB 3.8 新特性

InfluxDB 3.8 和 InfluxDB 3 Explorer 1.6 的主要增强功能。

查看博客文章

InfluxDB 3.8 现已适用于 Core 和 Enterprise 版本,同时发布了 InfluxDB 3 Explorer UI 的 1.6 版本。本次发布着重于操作成熟度,以及如何更轻松地部署、管理和可靠地运行 InfluxDB。

更多信息,请查看

InfluxDB Docker 的 latest 标签将指向 InfluxDB 3 Core

在 **2026 年 2 月 3 日**,InfluxDB Docker 镜像的 latest 标签将指向 InfluxDB 3 Core。为避免意外升级,请在您的 Docker 部署中使用特定的版本标签。

如果使用 Docker 来安装和运行 InfluxDB,latest 标签将指向 InfluxDB 3 Core。为避免意外升级,请在您的 Docker 部署中使用特定的版本标签。例如,如果使用 Docker 运行 InfluxDB v2,请将 latest 版本标签替换为 Docker pull 命令中的特定版本标签 — 例如

docker pull influxdb:2