企业官网建设流程全解析

设备端MQTT开发：Linux/嵌入式设备传感器数据上报实操

搭好了 Broker，定好了 Topic，接下来就是让设备真正"开口说话"。本文从 Python 到 C 语言，覆盖主流设备端开发方案，带你跑通传感器数据采集到 MQTT 上报的全流程。

一、设备端MQTT开发选型

不同硬件平台适合不同的开发语言和 MQTT 库：

选型建议：开发和验证阶段用 Python 快速跑通流程；量产部署到资源受限的嵌入式设备时，用 C 语言重写核心逻辑。两者可以并行——Python 脚本跑在树莓派上做网关，C 程序跑在 STM32 上做采集节点。

二、Python paho-mqtt完整实现

2.1 安装

pipinstallpaho-mqtt

2.2 核心API速览

importpaho.mqtt.clientasmqtt# 创建客户端client=mqtt.Client(client_id="farm01_gh03_sensor_01")# 设置认证（如果Broker开启了认证）client.username_pw_set("device_user","device_password")# 设置遗嘱消息（掉线时Broker自动发布）client.will_set("agriculture/farm01/greenhouse03/sensor/01/status",payload='{"status": "offline"}',qos=1)# 回调函数client.on_connect=on_connect# 连接成功回调client.on_message=on_message# 收到消息回调client.on_disconnect=on_disconnect# 断开连接回调# 连接Brokerclient.connect("192.168.1.100",1883,keepalive=60)# 订阅Topicclient.subscribe("agriculture/farm01/greenhouse03/controller/+/command")# 发布消息client.publish("agriculture/farm01/greenhouse03/sensor/temperature",payload='{"temp": 25.3, "ts": 1704067200}',qos=1)# 启动网络循环（保持心跳 + 处理回调）client.loop_start()# 后台线程运行# 或 client.loop_forever() # 阻塞运行

2.3 回调函数说明

paho-mqtt 采用异步回调模式，核心回调函数有三个：

• on_connect：连接 Broker 成功后触发，通常在这里执行订阅操作。
• on_message：收到订阅消息后触发，在这里处理收到的指令或数据。
• on_disconnect：连接断开后触发，在这里记录日志和触发重连。

三、传感器数据采集到MQTT发布全流程

下面是一个完整的智慧农业传感器上报程序，涵盖 DHT22 温湿度传感器数据采集、JSON 封装、MQTT 发布、定时循环的全流程。

#!/usr/bin/env python3""" 智慧农业传感器数据上报程序 硬件：Raspberry Pi + DHT22温湿度传感器 + 土壤湿度传感器 功能：定时采集传感器数据，通过MQTT上报到服务器 """importjsonimporttimeimportloggingimportpaho.mqtt.clientasmqttfromdatetimeimportdatetime# 尝试导入传感器库（开发环境可能没有硬件，用模拟数据）try:importAdafruit_DHT SENSOR_AVAILABLE=TrueexceptImportError:SENSOR_AVAILABLE=Falselogging.warning("传感器库未安装，将使用模拟数据")# ==================== 配置 ====================BROKER_HOST="192.168.1.100"BROKER_PORT=1883KEEPALIVE=60FARM_ID="farm01"GREENHOUSE_ID="greenhouse03"DEVICE_ID="sensor_01"# Topic定义TOPIC_TEMP=f"agriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/temperature"TOPIC_HUMIDITY=f"agriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/humidity"TOPIC_SOIL=f"agriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/soil_moisture"TOPIC_STATUS=f"agriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/{DEVICE_ID}/status"# 采集间隔（秒）REPORT_INTERVAL=30# ==================== MQTT回调 ====================defon_connect(client,userdata,flags,rc):"""连接成功回调"""ifrc==0:logging.info("连接Broker成功")# 上线通知client.publish(TOPIC_STATUS,json.dumps({"status":"online","ip":get_local_ip(),"ts":datetime.now().isoformat()}),qos=1)else:logging.error(f"连接失败，错误码:{rc}")defon_disconnect(client,userdata,rc):"""断开连接回调"""ifrc!=0:logging.warning(f"意外断开连接，rc={rc}，将自动重连")else:logging.info("主动断开连接")defon_message(client,userdata,msg):"""收到消息回调（处理下发指令）"""logging.info(f"收到消息: topic={msg.topic}, payload={msg.payload.decode()}")# ==================== 传感器采集 ====================defread_dht22():"""读取DHT22温湿度传感器"""ifSENSOR_AVAILABLE:humidity,temperature=Adafruit_DHT.read_retry(Adafruit_DHT.DHT22,4)returntemperature,humidityelse:# 模拟数据importrandomreturnround(random.uniform(20,30),1),round(random.uniform(50,70),1)defread_soil_moisture():"""读取土壤湿度（模拟值0-100%）"""ifSENSOR_AVAILABLE:# 实际项目中通过ADC读取模拟电压值并转换passimportrandomreturnround(random.uniform(30,60),1)defget_local_ip():"""获取本机IP地址"""importsockettry:s=socket.socket(socket.AF_INET,socket.SOCK_DGRAM)s.connect(("8.8.8.8",80))ip=s.getsockname()[0]s.close()returnipexceptException:return"unknown"# ==================== 主循环 ====================defmain():logging.basicConfig(level=logging.INFO,format='%(asctime)s [%(levelname)s] %(message)s')# 初始化MQTT客户端client=mqtt.Client(client_id=f"{FARM_ID}_{GREENHOUSE_ID}_{DEVICE_ID}")client.username_pw_set("device_user","device_password")# 遗嘱消息：设备掉线时自动发布client.will_set(TOPIC_STATUS,json.dumps({"status":"offline","ts":datetime.now().isoformat()}),qos=1)client.on_connect=on_connect client.on_disconnect=on_disconnect client.on_message=on_message# 启用自动重连（指数退避）client.reconnect_delay_set(min_delay=1,max_delay=60)# 连接Brokerclient.connect(BROKER_HOST,BROKER_PORT,KEEPALIVE)client.loop_start()# 后台线程处理网络IOlogging.info("传感器数据上报程序启动")# 定时采集上报循环try:whileTrue:# 1. 采集传感器数据temp,hum=read_dht22()soil=read_soil_moisture()ts=datetime.now().isoformat()# 2. 封装JSON并发布client.publish(TOPIC_TEMP,json.dumps({"value":temp,"unit":"C","ts":ts}),qos=1)client.publish(TOPIC_HUMIDITY,json.dumps({"value":hum,"unit":"%","ts":ts}),qos=1)client.publish(TOPIC_SOIL,json.dumps({"value":soil,"unit":"%","ts":ts}),qos=1)logging.info(f"数据上报完成: temp={temp}C, hum={hum}%, soil={soil}%")# 3. 等待下次采集time.sleep(REPORT_INTERVAL)exceptKeyboardInterrupt:logging.info("程序手动停止")finally:# 主动断开前发布离线状态client.publish(TOPIC_STATUS,json.dumps({"status":"offline","ts":datetime.now().isoformat()}),qos=1)client.loop_stop()client.disconnect()if__name__=="__main__":main()

代码要点解析：

1. 遗嘱消息：连接时设置了will_set，设备意外断线时 Broker 自动发布离线状态，服务端无需额外检测。
2. 自动重连：reconnect_delay_set配置了指数退避重连（1s→60s），网络恢复后自动重连。
3. QoS 1：传感器数据用 QoS 1 发布，保证消息至少送达一次，不丢数据。
4. 业务数据即心跳：上报周期 30s < Keep Alive 60s，每次 PUBLISH 自动充当心跳，无需额外发送 PINGREQ。

四、嵌入式Linux设备MQTT客户端

在 ARM 平台（如瑞芯微 RK3566）上，通常用 C 语言开发，配合 paho.mqtt.c 库。

4.1 交叉编译paho.mqtt.c

# 下载源码gitclone https://github.com/eclipse/paho.mqtt.c.gitcdpaho.mqtt.c# 交叉编译（指定ARM工具链）mkdirbuild&&cdbuild cmake..\-DCMAKE_C_COMPILER=arm-linux-gnueabihf-gcc\-DCMAKE_CXX_COMPILER=arm-linux-gnueabihf-g++\-DPAHO_WITH_SSL=OFF\-DPAHO_BUILD_STATIC=ON\-DPAHO_BUILD_SHARED=OFF\-DCMAKE_INSTALL_PREFIX=/opt/paho-armmake&&makeinstall

编译后将.a静态库和头文件拷贝到交叉编译工具链的搜索路径中。

4.2 C语言MQTT客户端核心框架

#include<stdio.h>#include<stdlib.h>#include<string.h>#include<unistd.h>#include"MQTTClient.h"#defineBROKER_ADDRESS"tcp://192.168.1.100:1883"#defineCLIENT_ID"farm01_greenhouse03_sensor_01"#defineTOPIC_TEMP"agriculture/farm01/greenhouse03/sensor/temperature"#defineQOS1#defineTIMEOUT10000L// 毫秒intmain(){MQTTClient client;MQTTClient_connectOptions conn_opts=MQTTClient_connectOptions_initializer;MQTTClient_message pubmsg=MQTTClient_message_initializer;intrc;// 1. 创建客户端if((rc=MQTTClient_create(&client,BROKER_ADDRESS,CLIENT_ID,MQTTCLIENT_PERSISTENCE_NONE,NULL))!=MQTTCLIENT_SUCCESS){printf("创建客户端失败, rc=%d\n",rc);return-1;}// 2. 配置连接参数conn_opts.keepAliveInterval=60;conn_opts.cleansession=1;conn_opts.username="device_user";conn_opts.password="device_password";conn_opts.connectTimeout=10;// 3. 连接Brokerif((rc=MQTTClient_connect(client,&conn_opts))!=MQTTCLIENT_SUCCESS){printf("连接失败, rc=%d\n",rc);return-1;}printf("连接成功\n");// 4. 循环发布传感器数据charpayload[256];while(1){// 读取传感器（伪代码）floattemp=read_temperature();floathum=read_humidity();// 封装JSONsnprintf(payload,sizeof(payload),"{\"temp\": %.1f, \"hum\": %.1f, \"ts\": %ld}",temp,hum,time(NULL));// 发布消息pubmsg.payload=payload;pubmsg.payloadlen=(int)strlen(payload);pubmsg.qos=QOS;pubmsg.retained=0;MQTTClient_publishMessage(client,TOPIC_TEMP,&pubmsg,NULL);printf("已发布: %s\n",payload);sleep(30);// 30秒上报一次}// 5. 断开连接并释放资源MQTTClient_disconnect(client,TIMEOUT);MQTTClient_destroy(&client);return0;}

4.3 守护进程方式运行

在嵌入式 Linux 上，用 systemd service 让程序开机自启、崩溃自动重启：

# /etc/systemd/system/agri-sensor.service [Unit] Description=Smart Agriculture Sensor Client After=network.target [Service] Type=simple ExecStart=/usr/local/bin/agri-sensor Restart=always RestartSec=5 User=root [Install] WantedBy=multi-user.target

systemctlenableagri-sensor# 开机自启systemctl start agri-sensor# 立即启动journalctl-uagri-sensor-f# 查看日志

五、设备端常见问题

5.1 内存泄漏

C 语言中最常见的问题。MQTTClient_message和MQTTClient_deliveryToken等对象用完必须释放。尤其是on_message回调中收到的消息体，处理完后要调用MQTTClient_freeMessage释放，否则长时间运行内存必爆。

5.2 网络切换恢复

设备从 Wi-Fi 切换到 4G（或反过来）时，TCP 连接会中断。paho-mqtt 的reconnect_delay_set可以处理自动重连，但 C 版本需要自行实现重连逻辑：

// 断线重连逻辑while(MQTTClient_isConnected(client)==0){rc=MQTTClient_connect(client,&conn_opts);if(rc!=MQTTCLIENT_SUCCESS){printf("重连失败, rc=%d, %d秒后重试\n",rc,retry_delay);sleep(retry_delay);retry_delay=(retry_delay<60)?retry_delay*2:60;// 指数退避}else{printf("重连成功\n");retry_delay=1;// 重置退避}}

5.3 消息队列积压

弱网环境下，publish可能长时间发不出去。QoS 1/2 的消息会在客户端本地排队，积压过多会耗尽内存。解决方案：

• 限制本地队列长度，超出后丢弃最旧的消息（传感器数据丢了无所谓，最新的最重要）
• 弱网时降低上报频率（比如从 30s 降到 60s）
• 关键数据用 QoS 1，非关键数据用 QoS 0

六、设备端MQTT最佳实践Checklist

• Client ID 全局唯一（农场+大棚+设备编号组合）
• 设置了遗嘱消息（LWT），掉线自动通知
• Keep Alive 值根据网络环境设定（弱网 30s，稳定网络 60s）
• 启用自动重连，指数退避（1s→60s）
• 传感器数据用 QoS 1，保证不丢
• 业务上报周期 ≤ Keep Alive，利用业务数据顶替心跳
• JSON 格式统一，包含 timestamp 字段
• C 语言程序注意释放 MQTT 对象，避免内存泄漏
• 嵌入式设备用 systemd 守护进程运行，崩溃自动重启
• 限制本地消息队列长度，防止弱网积压