智慧农业数据业务落地温湿度/土壤EC/PH值时序数据入库作者黒漂技术佬前面的文章把消息通路搭好了——MQTT 接到设备数据SpringBoot 接入微服务分发。但数据最终要去哪总不能光在消息队列里兜圈子吧。这篇文章聚焦「数据落地」——传感器数据怎么入库、存什么数据库、怎么查询、怎么管理数据生命周期。这是智慧农业平台真正产生业务价值的一步。一、智慧农业数据全景先看看一个大棚里到底有多少种数据在流动1.1 环境数据高频采集通常5~30秒一次传感器类型采集参数典型范围单位空气温湿度温度、湿度-20~60℃ / 0~100%℃ / %RH光照传感器光照强度0~200000LuxCO2传感器CO2浓度0~5000ppm风速风向风速、风向0~30m/s1.2 土壤数据中频采集通常1~5分钟一次传感器类型采集参数说明土壤温湿度土壤温度、体积含水量埋在不同深度 (10cm/30cm/50cm)土壤EC值电导率反映土壤盐分浓度0~20 mS/cm土壤PH值酸碱度0~14多数作物适宜 5.5~7.5氮磷钾NPK养分含量速效氮/磷/钾mg/kgEC值Electrical Conductivity电导率是很多新手会忽略的重要指标。它反映的是土壤溶液中可溶性盐的浓度。EC值过高说明土壤盐分超标植物根系会「烧」掉——就像你把植物种在盐水里一样。水肥一体机施肥时尤其要监控 EC 值不然肥施多了比不施还糟糕。1.3 设备状态数据水泵开关状态、水肥机流量、卷帘机位置开了百分之多少、补光灯功率等。这些数据和传感器数据不同——它们通常是事件驱动的状态变了才上报而非固定频率采集。二、数据库技术选型不同数据放不同数据库这是数据库选型的第一原则。别指望一种数据库通吃所有场景——那和用一把螺丝刀修所有电器没什么区别。数据类型推荐数据库原因时序数据温湿度、EC等InfluxDB / TDengine专为时序优化写入快、压缩高、聚合查询强关系数据设备台账、告警记录MySQL / PostgreSQL事务支持、复杂关联查询日志数据MongoDB / ElasticsearchSchema灵活、全文检索重点说说时序数据库。普通关系型数据库存时序数据有两个致命缺陷1. 写入性能不足。1000 个传感器每秒各写一条数据一天就是 8640 万条。MySQL 的 BTree 索引在这种写入压力下会频繁分裂性能急剧下降。2. 存储浪费巨大。时序数据高度重复设备 ID、传感器类型等冗余严重。时序数据库用列式存储 差值编码能把数据压缩到原始大小的十分之一。选 InfluxDB 还是 TDengine简单说如果你的团队熟悉 SQLTDengine 更友好标准 SQL 超级表模型如果已经有 InfluxDB 运维经验或者对 Flux 查询语言不排斥InfluxDB 生态更成熟。本文以 InfluxDB 为例。三、MQTT消息到数据入库的完整链路MQTT消息到达 │ ▼ 解析 JSON → 提取字段 │ deviceId, sensorType, value, timestamp ▼ 数据清洗异常值剔除 │ 温度突变 10℃标记异常不入库 │ 湿度 100%明显错误丢弃 ▼ 构建数据模型 │ Point.measurement(sensor_data) │ .tag(device_id, s001) │ .tag(sensor_type, temperature) │ .addField(value, 28.5) ▼ 批量写入 InfluxDB关键代码实现ServicepublicclassSensorDataPersistenceService{AutowiredprivateInfluxDBinfluxDB;// 批量缓冲区攒够100条或者满5秒就写入privatefinalListPointbuffernewArrayList();privatelonglastFlushTimeSystem.currentTimeMillis();privatestaticfinalintBATCH_SIZE100;privatestaticfinallongFLUSH_INTERVAL5000;publicsynchronizedvoidwritePoint(Pointpoint){buffer.add(point);if(buffer.size()BATCH_SIZE||System.currentTimeMillis()-lastFlushTimeFLUSH_INTERVAL){flush();}}privatevoidflush(){if(buffer.isEmpty())return;try{influxDB.write(BATCH_SIZE,TimeUnit.SECONDS,buffer);buffer.clear();lastFlushTimeSystem.currentTimeMillis();}catch(Exceptione){log.error(写入 InfluxDB 失败,e);// 写入失败不丢数据保留在buffer中等待下一次flush}}}为什么要批量写而不是来一条写一条InfluxDB 写入操作是有开销的——每条 HTTP 请求都涉及网络往返。100 条数据一次写和 100 次各写一条性能差距可能是 50 倍以上。这叫batch write批量写入是时序数据库的标准操作方式。四、InfluxDB 数据建模InfluxDB 的数据模型用measurement tag field timestamp来描述publicPointbuildSensorPoint(SensorDatadata){returnPoint.measurement(sensor_data)// measurement 表名.time(data.getTimestamp(),TimeUnit.MILLISECONDS)// 时间戳.tag(farm_id,data.getFarmId())// tag 索引列.tag(greenhouse_id,data.getGreenhouseId())// tag支持group by.tag(device_id,data.getDeviceId()).tag(sensor_type,data.getSensorType())// temperature/humidity/ec/ph.addField(value,data.getValue())// field 值列不建索引.addField(unit,data.getUnit()).build();}Tag 和 Field 的区别很重要Tag会被索引用于WHERE和GROUP BY。适合设备 ID、传感器类型这类基数不高的字符串。Field不建索引存具体数值。适合温度值、湿度值这类持续变化的数值。一个常见错误是把传感器数值也设为 Tag——这会导致 InfluxDB 索引急剧膨胀查询反而变慢。Tag 是用来「找数据」的Field 才是真正的「数据」。五、数据查询API设计有了数据还得给前端大屏、管理后台提供查询接口RestControllerRequestMapping(/api/sensor)publicclassSensorDataController{AutowiredprivateInfluxDBinfluxDB;/** * 查询某大棚某传感器的时间段温度曲线 * GET /api/sensor/query?greenhouseIdgh01sensorTypetemperaturestartxxxendxxx */GetMapping(/query)publicListDataPointquery(RequestParamStringgreenhouseId,RequestParamStringsensorType,RequestParamlongstart,RequestParamlongend){QueryquerynewQuery(String.format(SELECT MEAN(\value\) as \value\ FROM \sensor_data\ WHERE \greenhouse_id\%s AND \sensor_type\%s AND time %dms AND time %dms GROUP BY time(1m) fill(previous),greenhouseId,sensorType,start,end),smart_agriculture// 数据库名);QueryResultresultinfluxDB.query(query);returnconvertToDataPoints(result);}/** * 最近1小时各传感器均值 */GetMapping(/latest-hour-avg)publicMapString,DoublelatestHourAvg(RequestParamStringgreenhouseId){// 查询最近的温度均值和湿度均值QuerytempQuerynewQuery(String.format(SELECT MEAN(\value\) FROM \sensor_data\ WHERE \greenhouse_id\%s AND \sensor_type\temperature AND time now() - 1h,greenhouseId),smart_agriculture);// 类似查询湿度...// 返回 {temperature: 26.5, humidity: 72.3}returnresult;}}GROUP BY time(1m)是按 1 分钟聚合这在画折线图时非常有用——原始数据可能每秒都有但图表只需要分钟级别的粒度。fill(previous)是指定聚合窗口没有数据时用上一个窗口的值填充避免折线图出现断崖。六、数据保留策略全存是不可能全存的每天几千万条传感器数据硬盘再大也扛不住。需要制定分层保留策略-- InfluxDB 保留策略配置-- 原始数据保留30天CREATERETENTION POLICYraw_30dONsmart_agricultureDURATION30dREPLICATION1DEFAULT;-- 小时聚合数据保留1年通过连续查询自动生成CREATECONTINUOUS QUERYcq_hourly_avgONsmart_agricultureBEGINSELECTMEAN(value)ASvalueINTOsmart_agriculture.hourly.sensor_data_hourlyFROMsmart_agriculture.raw_30d.sensor_dataGROUPBYtime(1h),*END;-- 日聚合数据永久保留CREATERETENTION POLICYforeverONsmart_agricultureDURATION INFREPLICATION1;这就像视频监控——原始录像存 30 天关键片段永久保留。30 天后一秒一次的温度数据被聚合为 1 小时均值、1 日均值。查询去年今天的温度看日聚合就够了没必要翻秒级原始数据。连续查询Continuous Query是 InfluxDB 的杀手级特性。它自动按固定频率比如每小时执行聚合 SQL把结果写入另一个 measurement。你完全不需要写定时任务代码。总结智慧农业数据落地的关键决策就三个时序数据走时序库——别用 MySQL 存传感器数据那是自讨苦吃批量写入——100 条一写比 100 次各写 1 条快一个数量级分层保留——原始数据定时清聚合数据长期存数据入库不是终点而是业务价值的起点——有了数据你才能画曲线、做分析、触发告警、训练模型。