MQTT 物联网通信100 个传感器的数据怎么传到云端上一篇 ESP32 固件写好了但数据的另一端——云端消息中间件怎么搭这篇从零部署 EMQX设计 topic 规范和 JSON 格式加上设备上下线监控和断网重连一套生产可用的物联网通信就出来了。为什么是 MQTT EMQX物联网通信有几个硬需求低带宽农田里 4G 信号可能只有两格包要小低功耗设备电池供电不能维持长连接轮询双向不只是上报数据还要下发指令开阀、关泵大规模大棚多了上百个设备要同时在线MQTT 的发布/订阅模型天生契合设备 publish 数据到 broker后端 subscribe 接收存储控制指令反向 subscribe publish。EMQX 5.x 开源版免费单节点能扛 10 万连接对我们来说完全够用。EMQX 部署——一条 Docker 命令dockerrun-d\--nameemqx\--restartunless-stopped\-p1883:1883\# MQTT 端口-p8083:8083\# WebSocket 端口小程序用-p18083:18083\# Dashboard 管理界面emqx/emqx:5.7.0访问http://你的服务器IP:18083默认账号admin/public。进去后第一件事改密码并在「访问控制 → 认证」里创建一个 MQTT 用户比如farm_devices / 一个复杂密码所有 ESP32 用这个用户名密码连接。Topic 设计——房子的门牌号MQTT 没有数据库topic 就是唯一标识。好的 topic 设计让你查问题一目了然坏的 topic 设计让你半年后对着 Dashboard 一脸懵。规范{业务域}/{场地}/{设备类型}/{设备ID}/{消息类型}拆解层级示例值含义业务域farm农场业务区别于以后可能的homefactory场地greenhouse_01大棚编号设备类型sensor/actuator/camera设备角色设备IDesp32_a1b2c3唯一设备标识可以用 MAC 地址后 6 位消息类型data/status/cmd/alarm消息用途完整示例farm/greenhouse_01/sensor/esp32_a1b2c3/data ← 设备上报数据JSON farm/greenhouse_01/sensor/esp32_a1b2c3/status ← 设备状态在线/电池/信号 farm/greenhouse_01/actuator/pump_01/cmd ← 控制指令平台→设备 farm/greenhouse_01/sensor//data ← 通配符该棚所有传感器数据 farm////alarm ← 通配符整个农场所有告警JSON 数据格式——统一方言每个 sensor 上报的 data JSON 必须统一字段名不然后端解析要写一堆 if-else。传感器数据上报设备 → 平台{dev:esp32_a1b2c3,ts:1718000000,ver:1.0.0,data:{air_temp:26.5,air_humidity:68.2,soil_temp:22.1,soil_moisture:35.0,light:42000},battery:3.82,rssi:-65}字段说明dev设备 ID必须与 topic 中一致便于日志追踪tsUnix 时间戳秒ESP32 从 NTP 同步ver固件版本OTA 升级时判断data传感器读数key 固定battery电池电压V 3.3V 告警rssiWiFi 信号强度控制指令下发平台 → 设备{cmd:irrigate,seq:1823,params:{duration:300,valve:1}}seq是命令序列号递增。设备收到后回一个 ACK带相同 seq平台就知道这条命令确实送达了。没有 ACK 机制的命令下发就是丢硬币——丢了也不知道。ACK 回复设备 → 平台{type:ack,seq:1823,result:ok,msg:}设备订阅farm/greenhouse_01/actuator//cmd平台 publish 到具体设备的 cmd topic。设备收到后执行执行完 publish ACK 到.../ack。设备上下线管理——谁在线、谁掉线了MQTT 有个开箱即用的好东西遗嘱消息Last Will。设备连接时声明一个遗嘱 topic一旦异常断线心跳超时broker 自动帮你发布一条遗嘱消息。ESP32 端连接时设置mqtt.connect(MQTT_CLIENT_ID,MQTT_USER,MQTT_PASS,farm/greenhouse_01/sensor/esp32_a1b2c3/status,// 遗嘱 topic1,// QoStrue,// retain{\online\:false,\ts\:1718000000}// 遗嘱 payload);设备上线时自己 publish 一条{online:true}并设置 retaintrue新订阅者也能立刻拿到最新状态。后端监听Spring Boot 侧后端订阅farm////status收到消息后更新设备在线状态到 MySQL触发告警逻辑MqttListener(topicfarm////status)publicvoidonDeviceStatus(Stringtopic,Stringpayload){DeviceStatusstatusJSON.parseObject(payload,DeviceStatus.class);deviceService.updateOnlineStatus(status.getDev(),status.isOnline());if(!status.isOnline()){alarmService.trigger(设备离线: status.getDev());}}断网重连——农村的真实日常农村 WiFi 断网三件套下雨、打雷、光猫被拔要插电饭锅。ESP32 的断网重连逻辑必须健壮。// 带指数退避的重连voidconnectMQTT(){intretry0;intdelayMs1000;while(!mqtt.connected()){Serial.printf(MQTT 连接中 (第%d次)...,retry1);if(mqtt.connect(MQTT_CLIENT_ID,MQTT_USER,MQTT_PASS,statusTopic,1,true,willMsg)){Serial.println(成功);mqtt.publish(statusTopic,{\online\:true},true);mqtt.subscribe(cmdTopic);// 重新订阅控制指令return;}Serial.printf(失败, rc%d\n,mqtt.state());retry;if(retry5){delayMs1000*(1retry);// 1s, 2s, 4s, 8s, 16s}else{delayMs60000;// 5 次后每 60s 重试}delay(delayMs);}}加上本地缓存——MQTT 连不上时把数据存到 SPIFFS 文件网络恢复后补传voidloop(){SensorData datacollectAll();if(mqtt.connected()){// 先补传本地缓存flushLocalCache();// 再发当前数据publishData(data);}else{// 存到本地文件appendToLocalCache(data);}delay(300000);}flushLocalCache()和appendToLocalCache()用 ESP32 的 SPIFFS 或 LittleFS 实现逻辑不复杂每条数据一行 JSON补传时逐行读取 publish发完清空文件。带宽和流量估算一条 data JSON 大约 120 字节MQTT 协议头约 10 字节TCP/IP 头约 40 字节总计约 170 字节/条。5 分钟一条一天 288 条约 48KB/天1.4MB/月。加上心跳 ACK一个设备一个月不超过 3MB。100 个设备也就 300MB/月——即使插个 4G 流量卡10 块钱的物联网套餐都够用。快速验证装好 EMQX 后开一个 MQTT 客户端工具MQTTX 好用免费连上 broker订阅farm/#。然后用 ESP32 烧录上篇代码你应该立刻在工具里看到数据涌入。farm/greenhouse_01/sensor/esp32_a1b2c3/data → {dev:esp32...,ts:17180...} farm/greenhouse_01/sensor/esp32_a1b2c3/status → {online:true}如果看不到排查顺序ESP32 WiFi 连上了吗 → MQTT 密码对了吗 → 防火墙 1883 端口开了吗 → topic 拼写有误吗。下一篇《计算机视觉在农业的应用作物识别 病虫害检测实战》——用 YOLOv8 训练一个能认出 10 种蔬菜 20 种病害的模型部署到 RK3588 上。