3、数据采集基础：MQTT协议简介、使用paho-mqtt订阅设备Topic、数据格式解析（JSON/Protobuf）

好，咱们进入第三章节。前面两章我们把基线是什么、为什么要建聊清楚了。从这章开始，要动真格的了——数据采集。

你想想看，IoT设备的行为基线，说白了就是“数据驱动的画像”。没有数据，一切都是空谈。而IoT世界里，数据怎么从设备端跑到服务器？MQTT协议是当之无愧的主力。今天我就带你把它扒开看看，再手把手把订阅和解析的代码跑通。

本章核心目标：理解MQTT为什么是IoT的事实标准，掌握用paho-mqtt订阅设备Topic的方法，并能解析JSON和Protobuf两种主流数据格式。

第三章知识体系：MQTT数据采集与解析 第三章 知识体系：MQTT数据采集与解析 MQTT 协议核心 协议基础与原理 paho-mqtt 订阅实战 数据格式解析 发布/订阅模型 QoS 等级 on_connect / on_message JSON vs Protobuf 💡 数据采集是基线的起点 协议选型 → 订阅实现 → 格式解析，缺一不可

3.1 MQTT协议：为什么IoT世界都在用？

MQTT全称是Message Queuing Telemetry Transport，消息队列遥测传输。名字挺长，但核心思想很简单：轻量、低带宽、支持不可靠网络。

我记得2016年做第一个智能路灯项目时，设备端用的是2G模块，网络动不动就断。HTTP轮询？根本扛不住。后来换成MQTT，问题迎刃而解。说白了，MQTT就是为这种“设备多、网络差、功耗敏感”的场景量身定做的。

核心概念：发布/订阅模型

MQTT不像HTTP那样是“客户端请求-服务器响应”的模式。它用的是发布/订阅模型。有三个角色：

  • Broker（代理）：消息中转站，类似邮局。所有消息都经过它。
  • Publisher（发布者）：发送消息的一方，通常是设备。
  • Subscriber（订阅者）：接收消息的一方，通常是后端服务。

设备把数据发到Broker的一个“Topic”上，后端订阅这个Topic就能收到。设备之间不需要直接通信，解耦得干干净净。

我的习惯：Topic命名一定要规范。我一般用 project/location/device_type/device_id/data_type 这种层级结构。比如 smart_factory/zone_a/temperature_sensor/001/temp。后期做基线分析时，通配符订阅非常方便。

QoS等级：消息可靠性的三个档位

MQTT定义了三个QoS（服务质量）等级。我在项目中踩过坑，这里给你说清楚：

QoS等级 含义 可靠性 适用场景
0 最多一次（At most once） 最低，可能丢消息 传感器高频数据，丢几条无所谓
1 至少一次（At least once） 中等，可能重复 状态上报，允许重复但不可丢失
2 恰好一次（Exactly once） 最高，开销最大 控制指令、计费数据

我曾经踩过的坑：做智能门锁项目时，所有消息都用了QoS 2。结果设备端CPU和带宽双双爆表。后来分析发现，门锁状态变化用QoS 1就够了，只有开锁指令才需要QoS 2。别盲目追求高可靠性，要根据业务场景选。

3.2 使用paho-mqtt订阅设备Topic

理论说完了，上代码。Python生态里，paho-mqtt是最成熟的MQTT客户端库。我几乎每个IoT项目都用它。

安装与基础连接

pip install paho-mqtt

下面是一个最基础的订阅示例。我会把关键点都标出来：

import paho.mqtt.client as mqtt
import json

# 1. 连接回调：Broker响应后触发
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("连接成功！")
        # 订阅Topic，QoS设为1
        client.subscribe("smart_factory/+/temperature_sensor/+/temp", qos=1)
    else:
        print(f"连接失败，返回码：{rc}")

# 2. 消息回调：收到消息时触发
def on_message(client, userdata, msg):
    print(f"收到消息：Topic={msg.topic}, QoS={msg.qos}")
    # 原始数据是bytes，先解码
    payload = msg.payload.decode('utf-8')
    print(f"原始数据：{payload}")

# 3. 创建客户端实例
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

# 4. 连接Broker（替换为你的Broker地址）
client.connect("192.168.1.100", 1883, 60)

# 5. 启动网络循环（阻塞方式）
client.loop_forever()

这段代码里，有几点我想特别强调：

  • 通配符 + 和 #：+匹配单层，#匹配多层。上面例子用 + 匹配了任意区域和任意设备ID，非常灵活。
  • 回调函数是核心：on_connect 和 on_message 是MQTT编程的入口。所有业务逻辑都从这里开始。
  • QoS选择：订阅时指定的QoS是“订阅者期望的最大QoS”。发布者可以发更高QoS的消息，但Broker会降级到订阅者指定的等级。

我的建议：生产环境中，loop_forever() 会阻塞主线程。我通常用 client.loop_start() 启动后台线程，这样主线程还能做其他事情，比如数据入库或模型计算。

3.3 数据格式解析：JSON与Protobuf

数据拿到手了，但设备发来的原始数据通常是二进制或字符串。你得把它解析成结构化数据，才能做基线分析。

IoT领域最常用的两种格式：JSON和Protobuf。一个简单直观，一个高效紧凑。我两个都用，看场景选。

JSON解析：简单但“胖”

JSON的好处是人类可读，调试方便。坏处是体积大，解析开销高。

假设设备上报的数据是这样的：

{
  "device_id": "temp_001",
  "timestamp": 1696000000,
  "temperature": 25.6,
  "humidity": 68.3,
  "battery": 3.7
}

解析代码很简单：

def on_message(client, userdata, msg):
    try:
        # 解析JSON
        data = json.loads(msg.payload.decode('utf-8'))
        device_id = data.get("device_id")
        temp = data.get("temperature")
        hum = data.get("humidity")
        print(f"设备 {device_id}：温度={temp}°C, 湿度={hum}%")
        # 这里就可以把数据传给基线模型了
    except json.JSONDecodeError as e:
        print(f"JSON解析失败：{e}")

注意：JSON解析一定要加try-except。我在项目中遇到过设备固件bug，偶尔会发畸形JSON。不加异常处理，整个订阅线程会挂掉。

Protobuf解析：高效但需要Schema

Protobuf是Google的序列化协议。它把数据压缩成二进制，体积只有JSON的1/3到1/5，解析速度也快得多。但代价是：你得先定义Schema（.proto文件），而且数据不可读。

先定义一个简单的proto文件：

syntax = "proto3";

message SensorData {
  string device_id = 1;
  uint64 timestamp = 2;
  float temperature = 3;
  float humidity = 4;
  float battery = 5;
}

然后用protoc编译成Python类：

protoc --python_out=. sensor.proto

在订阅代码里解析：

import sensor_pb2  # 编译生成的模块

def on_message(client, userdata, msg):
    # 直接解析二进制数据
    data = sensor_pb2.SensorData()
    data.ParseFromString(msg.payload)
    print(f"设备 {data.device_id}：温度={data.temperature}°C, 湿度={data.humidity}%")

看到区别了吗？Protobuf不需要decode成字符串，直接喂二进制就行。解析速度比JSON快一个数量级。

JSON vs Protobuf：怎么选？

对比维度 JSON Protobuf
可读性 ★★★★★ 人类可直接阅读 ★☆☆☆☆ 二进制，不可读
数据体积 ★★☆☆☆ 较大，有冗余 ★★★★★ 紧凑，压缩率高
解析速度 ★★★☆☆ 中等 ★★★★★ 极快
Schema要求 无，灵活但易出错 必须预定义，严格但安全
调试难度 低，curl就能看 高，需要工具解码
适用场景 原型验证、低频数据 大规模生产、带宽受限

我的经验：项目初期用JSON快速迭代，等协议稳定了再切Protobuf。我上一个项目就是这样做的——前两个月用JSON调试，第三个月统一切Protobuf，带宽直接降了60%。

3.4 实战：完整的订阅+解析流程

最后，我把今天的内容串成一个完整的示例。这个代码框架，你稍作修改就能用到自己的项目里：

import paho.mqtt.client as mqtt
import json
import sensor_pb2  # 假设Protobuf已编译

BROKER = "your.broker.com"
PORT = 1883
TOPIC = "factory/+/sensor/+/data"

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("已连接到Broker")
        client.subscribe(TOPIC, qos=1)
    else:
        print(f"连接失败，rc={rc}")

def on_message(client, userdata, msg):
    # 根据Topic后缀判断格式
    if msg.topic.endswith("/json"):
        try:
            data = json.loads(msg.payload.decode('utf-8'))
            process_json_data(data)
        except Exception as e:
            print(f"JSON解析异常：{e}")
    elif msg.topic.endswith("/pb"):
        try:
            data = sensor_pb2.SensorData()
            data.ParseFromString(msg.payload)
            process_protobuf_data(data)
        except Exception as e:
            print(f"Protobuf解析异常：{e}")
    else:
        print(f"未知数据格式：{msg.topic}")

def process_json_data(data):
    # 这里写你的基线处理逻辑
    print(f"[JSON] 设备 {data['device_id']} 温度={data['temperature']}")

def process_protobuf_data(data):
    # 这里写你的基线处理逻辑
    print(f"[Protobuf] 设备 {data.device_id} 温度={data.temperature}")

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, 60)
client.loop_forever()

重要提醒：生产环境中，on_message 回调里不要做耗时操作（比如写数据库、调用外部API）。否则会阻塞消息接收。我通常的做法是：回调里只做解析和入队列，后台另起线程池消费队列数据。

← 上一章 📖 返回目录 下一章 →