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 psutil

def 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 json

def 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 result

def 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_connect
    client.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 json

guance = 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 进行封装，上报至观测云平台进行统一的管理、可视化分析、告警等。