Spring Boot 农业物联网平台从 0 到 1 搭建传感器数据上了 MQTT摄像头跑起了 YOLO现在需要一个「大脑」把它们管起来。这篇从零搭一套 Spring Boot 3 物联网中台设备管理、TDengine 时序存储、告警规则引擎、ECharts 可视化。项目初始化# Spring Boot 3.2 JDK 17spring init\-dweb,websocket,mybatis,mysql,lombok,validation\-gcom.farmer\-afarm-iot-platform\farm-iot-platform额外依赖!-- TDengine 连接器 --dependencygroupIdcom.taosdata.jdbc/groupIdartifactIdtaos-jdbcdriver/artifactIdversion3.3.0/version/dependency!-- EMQX MQTT 客户端 --dependencygroupIdorg.eclipse.paho/groupIdartifactIdorg.eclipse.paho.client.mqttv3/artifactIdversion1.2.5/version/dependency!-- Sa-Token 权限 --dependencygroupIdcn.dev33/groupIdartifactIdsa-token-spring-boot3-starter/artifactIdversion1.38.0/version/dependency目录结构farm-iot-platform ├── controller/ → REST API 层 ├── service/ → 业务逻辑 ├── mapper/ → MyBatis Mapper ├── model/ │ ├── entity/ → MySQL 实体 │ └── dto/ → 数据传输对象 ├── mqtt/ → MQTT 消息处理 ├── alarm/ → 告警引擎 ├── websocket/ → 实时推送 └── config/ → 配置类设备管理CRUD 分组 标签MySQL 表设计CREATETABLEdevice(idBIGINTPRIMARYKEYAUTO_INCREMENT,device_idVARCHAR(64)NOTNULLUNIQUECOMMENT设备唯一ID, esp32_a1b2c3,nameVARCHAR(128)COMMENT设备名称, 大棚A东区传感器1,typeVARCHAR(32)NOTNULLCOMMENTsensor/actuator/camera,locationVARCHAR(256)COMMENT安装位置,group_idBIGINTCOMMENT所属分组,tagsVARCHAR(512)DEFAULTCOMMENT标签,逗号分隔,mqtt_topicVARCHAR(256)COMMENTMQTT topic 前缀,onlineTINYINTDEFAULT0COMMENT0离线 1在线,firmwareVARCHAR(32)COMMENT固件版本,latDOUBLECOMMENT纬度,lngDOUBLECOMMENT经度,create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP,update_timeDATETIMEDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,INDEXidx_device_id(device_id),INDEXidx_group_id(group_id))ENGINEInnoDBDEFAULTCHARSETutf8mb4;CREATETABLEdevice_group(idBIGINTPRIMARYKEYAUTO_INCREMENT,nameVARCHAR(64)NOTNULLCOMMENT分组名, 大棚A/果园B,parent_idBIGINTDEFAULT0COMMENT父分组ID,create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP)ENGINEInnoDB;实体 Mapper 用 MyBatis-Plus 一把梭TableName(device)publicclassDevice{TableId(typeIdType.AUTO)privateLongid;privateStringdeviceId;privateStringname;privateStringtype;// sensor / actuator / cameraprivateStringlocation;privateLonggroupId;privateStringtags;privateStringmqttTopic;privateBooleanonline;privateStringfirmware;}publicinterfaceDeviceMapperextendsBaseMapperDevice{}Service 层提供分页查询、按分组筛选、按标签搜索ServicepublicclassDeviceService{publicIPageDeviceVOpageDevices(DevicePageQueryquery){LambdaQueryWrapperDevicewrappernewLambdaQueryWrapper();// 关键搜索逻辑模糊匹配名称或设备IDif(StrUtil.isNotBlank(query.getKeyword())){wrapper.and(w-w.like(Device::getName,query.getKeyword()).or().eq(Device::getDeviceId,query.getKeyword()));}// 按分组筛选含子分组if(query.getGroupId()!null){ListLonggroupIdsgetGroupWithChildren(query.getGroupId());wrapper.in(Device::getGroupId,groupIds);}// 按标签筛选FIND_IN_SETif(StrUtil.isNotBlank(query.getTag())){wrapper.apply(FIND_IN_SET({0}, tags),query.getTag());}returndeviceMapper.selectPage(query.getPage(),wrapper).convert(this::toVO);}}TDengine 时序存储——写入快、查询快、压缩率高TDengine 专门为物联网场景设计写入性能是 InfluxDB 的 10 倍数据压缩率 10:1 以上。建超级表 自动子表-- 超级表定义数据结构和标签CREATESTABLEIFNOTEXISTSfarm.sensor_data(tsTIMESTAMP,air_tempFLOAT,air_humidityFLOAT,soil_tempFLOAT,soil_moistureFLOAT,lightINT,batteryFLOAT,rssiINT)TAGS(device_idBINARY(64),locationBINARY(256));-- 插入时自动创建子表子表名 d_设备IDINSERTINTOfarm.d_esp32_a1b2c3USINGfarm.sensor_data TAGS(esp32_a1b2c3,大棚A-东区-1号位)VALUES(NOW,26.5,68.2,22.1,35.0,42000,3.82,-65);Spring Boot 连接 TDengineConfigurationpublicclassTDengineConfig{BeanpublicJdbcTemplatetdengineJdbcTemplate(Value(${tdengine.url})Stringurl,Value(${tdengine.username})Stringusername,Value(${tdengine.password})Stringpassword){returnnewJdbcTemplate(newDriverManagerDataSource(url,username,password));}}application.ymltdengine:url:jdbc:TAOS://localhost:6030/farmusername:rootpassword:taosdata监听 MQTT 消息并写入 TDengineComponentpublicclassSensorDataListener{AutowiredprivateJdbcTemplatetdengine;MqttListener(topicfarm//sensor//data)publicvoidonSensorData(Stringtopic,Stringpayload){SensorDatadataJSON.parseObject(payload,SensorData.class);// 提取 sensor 之后的设备IDtopic 第 3 段StringdeviceIdtopic.split(/)[3];StringtableNamefarm.d_deviceId.replace(-,_);tdengine.update(INSERT INTO tableName VALUES (?, ?, ?, ?, ?, ?, ?, ?),data.getTs(),data.getAirTemp(),data.getAirHumidity(),data.getSoilTemp(),data.getSoilMoisture(),data.getLight(),data.getBattery(),data.getRssi());}}查询——TDengine 标准 SQL-- 最近 24 小时数据每 5 分钟一个点降采样SELECT_wstart,AVG(air_temp),AVG(air_humidity),AVG(soil_moisture)FROMfarm.sensor_dataWHEREtsNOW-1dINTERVAL(5m);-- 多设备对比大棚 A 和 B 过去 1 小时温度SELECTts,air_tempFROMfarm.d_esp32_a1b2c3WHEREtsNOW-1hUNIONALLSELECTts,air_tempFROMfarm.d_esp32_d4e5f6WHEREtsNOW-1h;告警规则引擎——不要整 DSL正则就够了告警配置存 MySQLCREATETABLEalarm_rule(idBIGINTPRIMARYKEYAUTO_INCREMENT,nameVARCHAR(128),device_idVARCHAR(64),-- NULL 所有设备fieldVARCHAR(32),-- air_temp / soil_moisture / battery / onlineoperatorVARCHAR(8),-- / / / !thresholdDOUBLE,durationINTDEFAULT0,-- 持续秒数0 立即触发levelVARCHAR(16)DEFAULTwarn,-- warn / error / criticalenabledTINYINTDEFAULT1,create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP);告警引擎——每次收到传感器数据后触发评估ServicepublicclassAlarmEngine{AutowiredprivateAlarmRuleMapperruleMapper;AutowiredprivateAlarmRecordMapperrecordMapper;AutowiredprivateWebSocketServicewsService;privateMapString,LocalDateTimeconditionMetSincenewConcurrentHashMap();publicvoidevaluate(SensorDatadata){ListAlarmRulerulesruleMapper.selectList(newLambdaQueryWrapperAlarmRule().eq(AlarmRule::getEnabled,true).and(w-w.isNull(AlarmRule::getDeviceId).or().eq(AlarmRule::getDeviceId,data.getDev())));for(AlarmRulerule:rules){doublevaluedata.getValueByField(rule.getField());booleantriggeredevaluateRule(rule.getOperator(),value,rule.getThreshold());StringruleKeyrule.getId()_data.getDev();if(triggered){LocalDateTimesinceconditionMetSince.get(ruleKey);if(sincenull){conditionMetSince.put(ruleKey,LocalDateTime.now());}elseif(ChronoUnit.SECONDS.between(since,LocalDateTime.now())rule.getDuration()){fireAlarm(rule,data,value);}}else{conditionMetSince.remove(ruleKey);}}}privatevoidfireAlarm(AlarmRulerule,SensorDatadata,doublevalue){// 记录告警AlarmRecordrecordnewAlarmRecord();record.setRuleId(rule.getId());record.setRuleName(rule.getName());record.setDeviceId(data.getDev());record.setField(rule.getField());record.setValue(value);record.setThreshold(rule.getThreshold());record.setLevel(rule.getLevel());recordMapper.insert(record);// 推送告警到前端wsService.push(MessageType.ALARM,record);}}告警类型示例规则条件级别土壤过干soil_moisture 20% 持续 10minwarn大棚高温air_temp 40℃error设备离线online falsecritical电池低电battery 3.4VwarnCO₂ 不足co2 300ppmwarnWebSocket 实时推送 ECharts 可视化后端 WebSocket 配置ConfigurationEnableWebSocketpublicclassWebSocketConfigimplementsWebSocketConfigurer{OverridepublicvoidregisterWebSocketHandlers(WebSocketHandlerRegistryregistry){registry.addHandler(newFarmWebSocketHandler(),/ws/farm).setAllowedOrigins(*);}}前端用 ECharts 动态刷新Vue 3 组件script setup import * as echarts from echarts; import { ref, onMounted, onUnmounted } from vue; const chartRef ref(null); let chart null; let ws null; onMounted(() { chart echarts.init(chartRef.value); chart.setOption({ title: { text: 大棚 A 实时温湿度 }, xAxis: { type: time }, yAxis: [ { type: value, name: 温度(℃) }, { type: value, name: 湿度(%) } ], series: [ { name: 温度, type: line, yAxisIndex: 0, data: [] }, { name: 湿度, type: line, yAxisIndex: 1, data: [] } ] }); // WebSocket 连接 ws new WebSocket(ws://your-server/ws/farm); ws.onmessage (event) { const data JSON.parse(event.data); const now new Date(data.ts * 1000); // 向图表追加数据点 chart.setOption({ series: [ { data: [...chart.getOption().series[0].data, [now, data.air_temp]] }, { data: [...chart.getOption().series[1].data, [now, data.air_humidity]] } ] }); }; }); onUnmounted(() { chart?.dispose(); ws?.close(); }); /script template div refchartRef stylewidth:100%;height:400px/div /template部署清单# 1. 阿里云 / 腾讯云 2C4G 服务器# 2. 安装 Docker# 3. 一条命令起三个容器dockerrun-d--nameemqx-p1883:1883-p18083:18083 emqx/emqx:5.7.0dockerrun-d--nametdengine-p6030:6030-p6041:6041 tdengine/tdengine:3.3.0dockerrun-d--namemysql-p3306:3306-eMYSQL_ROOT_PASSWORDxxx mysql:8.0# 4. 打包 Spring Boot 应用mvn clean package-DskipTests# 5. 运行java-jarfarm-iot-platform.jar--spring.profiles.activeprod下一篇《微信小程序农户手机上的「农场管家」》——UniApp 开发一个码农扫二维码就能看到自家大棚实时数据手指一点远程灌溉。