设备端MQTT开发：Linux/嵌入式设备传感器数据上报实操-北京尧图网络

设备端MQTT开发Linux/嵌入式设备传感器数据上报实操搭好了 Broker定好了 Topic接下来就是让设备真正开口说话。本文从 Python 到 C 语言覆盖主流设备端开发方案带你跑通传感器数据采集到 MQTT 上报的全流程。一、设备端MQTT开发选型不同硬件平台适合不同的开发语言和 MQTT 库语言MQTT库适用平台特点Pythonpaho-mqttRaspberry Pi / Linux工控机开发快调试方便适合原型和小规模部署C/Cpaho.mqtt.c / mosquittoSTM32 / RTOS / 嵌入式Linux性能高资源占用小适合量产设备Javaeclipse.pahoAndroid工控设备跨平台适合商用工控平板选型建议开发和验证阶段用 Python 快速跑通流程量产部署到资源受限的嵌入式设备时用 C 语言重写核心逻辑。两者可以并行——Python 脚本跑在树莓派上做网关C 程序跑在 STM32 上做采集节点。二、Python paho-mqtt完整实现2.1 安装pipinstallpaho-mqtt2.2 核心API速览importpaho.mqtt.clientasmqtt# 创建客户端clientmqtt.Client(client_idfarm01_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},qos1)# 回调函数client.on_connecton_connect# 连接成功回调client.on_messageon_message# 收到消息回调client.on_disconnecton_disconnect# 断开连接回调# 连接Brokerclient.connect(192.168.1.100,1883,keepalive60)# 订阅Topicclient.subscribe(agriculture/farm01/greenhouse03/controller//command)# 发布消息client.publish(agriculture/farm01/greenhouse03/sensor/temperature,payload{temp: 25.3, ts: 1704067200},qos1)# 启动网络循环保持心跳处理回调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_AVAILABLETrueexceptImportError:SENSOR_AVAILABLEFalselogging.warning(传感器库未安装将使用模拟数据)# 配置 BROKER_HOST192.168.1.100BROKER_PORT1883KEEPALIVE60FARM_IDfarm01GREENHOUSE_IDgreenhouse03DEVICE_IDsensor_01# Topic定义TOPIC_TEMPfagriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/temperatureTOPIC_HUMIDITYfagriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/humidityTOPIC_SOILfagriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/soil_moistureTOPIC_STATUSfagriculture/{FARM_ID}/{GREENHOUSE_ID}/sensor/{DEVICE_ID}/status# 采集间隔秒REPORT_INTERVAL30# MQTT回调 defon_connect(client,userdata,flags,rc):连接成功回调ifrc0:logging.info(连接Broker成功)# 上线通知client.publish(TOPIC_STATUS,json.dumps({status:online,ip:get_local_ip(),ts:datetime.now().isoformat()}),qos1)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,temperatureAdafruit_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:ssocket.socket(socket.AF_INET,socket.SOCK_DGRAM)s.connect((8.8.8.8,80))ips.getsockname()[0]s.close()returnipexceptException:returnunknown# 主循环 defmain():logging.basicConfig(levellogging.INFO,format%(asctime)s [%(levelname)s] %(message)s)# 初始化MQTT客户端clientmqtt.Client(client_idf{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()}),qos1)client.on_connecton_connect client.on_disconnecton_disconnect client.on_messageon_message# 启用自动重连指数退避client.reconnect_delay_set(min_delay1,max_delay60)# 连接Brokerclient.connect(BROKER_HOST,BROKER_PORT,KEEPALIVE)client.loop_start()# 后台线程处理网络IOlogging.info(传感器数据上报程序启动)# 定时采集上报循环try:whileTrue:# 1. 采集传感器数据temp,humread_dht22()soilread_soil_moisture()tsdatetime.now().isoformat()# 2. 封装JSON并发布client.publish(TOPIC_TEMP,json.dumps({value:temp,unit:C,ts:ts}),qos1)client.publish(TOPIC_HUMIDITY,json.dumps({value:hum,unit:%,ts:ts}),qos1)client.publish(TOPIC_SOIL,json.dumps({value:soil,unit:%,ts:ts}),qos1)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()}),qos1)client.loop_stop()client.disconnect()if__name____main__:main()代码要点解析遗嘱消息连接时设置了will_set设备意外断线时 Broker 自动发布离线状态服务端无需额外检测。自动重连reconnect_delay_set配置了指数退避重连1s→60s网络恢复后自动重连。QoS 1传感器数据用 QoS 1 发布保证消息至少送达一次不丢数据。业务数据即心跳上报周期 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工具链mkdirbuildcdbuild cmake..\-DCMAKE_C_COMPILERarm-linux-gnueabihf-gcc\-DCMAKE_CXX_COMPILERarm-linux-gnueabihf-g\-DPAHO_WITH_SSLOFF\-DPAHO_BUILD_STATICON\-DPAHO_BUILD_SHAREDOFF\-DCMAKE_INSTALL_PREFIX/opt/paho-armmakemakeinstall编译后将.a静态库和头文件拷贝到交叉编译工具链的搜索路径中。4.2 C语言MQTT客户端核心框架#includestdio.h#includestdlib.h#includestring.h#includeunistd.h#includeMQTTClient.h#defineBROKER_ADDRESStcp://192.168.1.100:1883#defineCLIENT_IDfarm01_greenhouse03_sensor_01#defineTOPIC_TEMPagriculture/farm01/greenhouse03/sensor/temperature#defineQOS1#defineTIMEOUT10000L// 毫秒intmain(){MQTTClient client;MQTTClient_connectOptions conn_optsMQTTClient_connectOptions_initializer;MQTTClient_message pubmsgMQTTClient_message_initializer;intrc;// 1. 创建客户端if((rcMQTTClient_create(client,BROKER_ADDRESS,CLIENT_ID,MQTTCLIENT_PERSISTENCE_NONE,NULL))!MQTTCLIENT_SUCCESS){printf(创建客户端失败, rc%d\n,rc);return-1;}// 2. 配置连接参数conn_opts.keepAliveInterval60;conn_opts.cleansession1;conn_opts.usernamedevice_user;conn_opts.passworddevice_password;conn_opts.connectTimeout10;// 3. 连接Brokerif((rcMQTTClient_connect(client,conn_opts))!MQTTCLIENT_SUCCESS){printf(连接失败, rc%d\n,rc);return-1;}printf(连接成功\n);// 4. 循环发布传感器数据charpayload[256];while(1){// 读取传感器伪代码floattempread_temperature();floathumread_humidity();// 封装JSONsnprintf(payload,sizeof(payload),{\temp\: %.1f, \hum\: %.1f, \ts\: %ld},temp,hum,time(NULL));// 发布消息pubmsg.payloadpayload;pubmsg.payloadlen(int)strlen(payload);pubmsg.qosQOS;pubmsg.retained0;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] DescriptionSmart Agriculture Sensor Client Afternetwork.target [Service] Typesimple ExecStart/usr/local/bin/agri-sensor Restartalways RestartSec5 Userroot [Install] WantedBymulti-user.targetsystemctlenableagri-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){rcMQTTClient_connect(client,conn_opts);if(rc!MQTTCLIENT_SUCCESS){printf(重连失败, rc%d, %d秒后重试\n,rc,retry_delay);sleep(retry_delay);retry_delay(retry_delay60)?retry_delay*2:60;// 指数退避}else{printf(重连成功\n);retry_delay1;// 重置退避}}5.3 消息队列积压弱网环境下publish可能长时间发不出去。QoS 1/2 的消息会在客户端本地排队积压过多会耗尽内存。解决方案限制本地队列长度超出后丢弃最旧的消息传感器数据丢了无所谓最新的最重要弱网时降低上报频率比如从 30s 降到 60s关键数据用 QoS 1非关键数据用 QoS 0六、设备端MQTT最佳实践ChecklistClient ID 全局唯一农场大棚设备编号组合设置了遗嘱消息LWT掉线自动通知Keep Alive 值根据网络环境设定弱网 30s稳定网络 60s启用自动重连指数退避1s→60s传感器数据用 QoS 1保证不丢业务上报周期 ≤ Keep Alive利用业务数据顶替心跳JSON 格式统一包含 timestamp 字段C 语言程序注意释放 MQTT 对象避免内存泄漏嵌入式设备用 systemd 守护进程运行崩溃自动重启限制本地消息队列长度防止弱网积压

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

相关推荐

STM32C5双ADC交错采样配置实战：从CubeMX到代码调通

MQTT主题设计规范：智慧农业设备分层Topic设计（设备/大棚/分组）

MQTT-EMQX服务部署：Docker搭建高可用MQTT服务器、参数调优

更多资讯

波导技术原理与应用全解析

小公司产品经理如何平衡售后与产品设计

SpringBoot+微信小程序构建电子书城实战

人工智能知识表示与逻辑推理能力校准指南

中文用户LoRA训练前置指南：SD标签生成工作流

K-Means+SVM栅格分区：为蚁群路径规划搜索空间减负

今日更新

AI辅助技术栈评估：提升研发决策效率60%

从生成Token到直接决策：Jev如何重塑AI Agent工程实践

32G统一内存跑大模型：内存账、TPS实测与端云分流

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？