当前位置：首页 > news >正文

MQTT协议下温度数据上报观测云最佳实践

news 来源：原创 2025/8/22 17:07:33

MQTT 介绍

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议）是一种轻量级的、基于发布/订阅模式的消息传输协议，专为低带宽、高延迟或不可靠的网络环境设计，广泛应用于物联网（IoT）、移动应用和分布式系统中，用于实现设备之间的高效通信。它通过减少数据传输量和简化通信流程，确保消息的可靠传输，同时支持多种服务质量（QoS）等级，以满足不同的业务需求。

观测云

观测云是一款专为 IT 工程师打造的全链路可观测产品，它集成了基础设施监控、应用程序性能监控和日志管理，为整个技术栈提供实时可观察性。这款产品能够帮助工程师全面了解端到端的用户体验追踪，了解应用内函数的每一次调用，以及全面监控云时代的基础设施。此外，观测云还具备快速发现系统安全风险的能力，为数字化时代提供安全保障。

Func 平台

观测云 Func 平台（DataFlux Func）是一个基于 Python 的函数计算与数据处理开发平台，旨在帮助用户快速搭建、管理和执行数据处理任务。它主要由 Server（提供 Web UI 和 API 接口）和 Worker（提供 Python 脚本执行环境）两部分组成。该平台支持多源数据对接，内置丰富的数据源连接器，可快速实现数据汇聚与处理。此外，Func 平台还具备定时任务管理、API 接口发布等功能，支持同步、异步和定时调用，极大简化了开发流程。通过与观测云的深度集成，用户可以利用其强大的可观测性能力，实现数据的实时处理、分析和可视化。

通过各类传感器接收设备数据，进行处理并通过 MQTT 协议上报，通过 Func 消费来自 MQTT 协议的数据并上报至观测云平台进行存储、展示、分析、预警等操作。

实战

场景：采集 Linux 主机温度，上报至观测云。
准备工作：EMQX，接收 MQTT 协议的数据并供 Client 端消费数据

采集温度

Sensors（传感器）采集温度是通过物理或化学原理将温度变化转换为电信号的过程。常见的温度传感器包括热敏电阻（NTC/PTC）、热电偶、热敏二极管和数字温度传感器（如DS18B20）。这些传感器通过感知环境温度的变化，输出与温度成正比或符合特定函数关系的电压、电流或数字信号。这些信号经过放大、滤波和模数转换后，可被微控制器或数据采集系统读取，从而实现对温度的实时监测和记录。

在 Shell 中执行 sensors 可以获取到当前 CPU、硬盘等温度信息。

通过脚本方式获取温度信息：

import psutildef get_system_temperatures():temps = psutil.sensors_temperatures()if not temps:return "No temperature sensors found."result = ""for chip, sensors in temps.items():result += f"{chip}:\n"for sensor in sensors:result += f"  {sensor.label or 'Sensor'}: {sensor.current}°C (high={sensor.high}, critical={sensor.critical})\n"return result# 调用函数并打印结果
system_temps = get_system_temperatures()print(system_temps)

说明：

psutil.sensors_temperatures() 返回系统中所有温度传感器的信息。
每个传感器有 current（当前温度）、high（高温警戒值）和 critical（临界温度）等属性。

上报到 EMQX

import psutil
import paho.mqtt.client as mqtt
import time
import jsondef get_system_temperatures():temps = psutil.sensors_temperatures()if not temps:return "No temperature sensors found."result = {}for chip, sensors in temps.items():result[chip] = []for sensor in sensors:if sensor.label is not None and sensor.label != "":result[chip].append({'host': "liurui",'label': sensor.label,'current': sensor.current,'high': sensor.high,'critical': sensor.critical})return resultdef on_connect(client, userdata, flags, rc):print(f"Connected with result code {rc}")def publish_temperatures(temps, broker, port, topic):client = mqtt.Client()client.on_connect = on_connectclient.connect(broker, port, 60)client.loop_start()# 打包所有传感器数据成一个JSON对象payload = json.dumps(temps)client.publish(topic, payload)print(f"Published: {payload}")client.loop_stop()if __name__ == "__main__":broker = "1.1.1.1"  # 替换为你的EMQX broker地址port = 1883  # 默认MQTT端口topic = "temperature"  # 替换为你想要发布的主题while True:system_temps = get_system_temperatures()if isinstance(system_temps, str):print(system_temps)else:publish_temperatures(system_temps, broker, port, topic)time.sleep(5)  # 每5秒采集并上报一次

申请观测云 API Key

登陆观测云控制台，点击菜单「管理」-「API Key 管理」，新建 API Key。

保存 Key ID 和 Key，后续 Func 平台需要用到。

Func 消费 MQTT 数据

1、新建脚本集

2、新建脚本

脚本内容如下：

import jsonguance = DFF.CONN('GuanceAPI')@DFF.API('Message Handler')
def message_handler(topic, message):print(f"Received message: {message} on topic {topic}")if topic == "temperature":parse_and_print_temperatures(message)def parse_and_print_temperatures(temps_json):# 解析 JSON 字符串为 Python 字典temps_dict = json.loads(temps_json)result = []# 遍历每个芯片for chip, sensors in temps_dict.items():print(f"Chip: {chip}")# 遍历每个传感器for sensor in sensors:result.append({'measurement': 'temperature','tags': {'host': sensor['host'],'chip': chip,'label': sensor['label']},'fields': {'current': sensor['current'],'high': sensor['high'],'critical': sensor['critical']}})uploadDataKit(result)def uploadDataKit(data):# 获取 DataKit 操作对象status_code, result = guance.dataway.write_by_category_many(category='metric', data=data)print(f"上报结果：{status_code}")

3、发布脚本

点击发布按钮进行发布。

4、创建观测云连接器

类型：观测云
ID： GuanceAPI

注意：ID 需要与脚本 guance = DFF.CONN('GuanceAPI') 的 ID 一致，其他字段按照实际情况填写。

5、创建 MQTT 连接器

类型为 MQTT Broker (v5.0)
填写 ID、主机、端口
选择主题以及对应主题消费的脚本
点击测试连通性，确保 MQTT 可以正常链接
点击保存即可

效果展示

数据上报至观测云后，可以通过仪表板使用以下 DQL 语句可以查看温度的趋势图。

M::`temperature`:(last(`current`)) BY `chip`, `label`

总结

通过观测云 Func 平台接收来自 MQTT 协议的指标、日志、链路等数据，并通过观测云提供的 API 进行封装，上报至观测云平台进行统一的管理、可视化分析、告警等。