智慧農業數據業務落地溫濕度/土壤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 條快一個數量級分層保留——原始數據定時清聚合數據長期存數據入庫不是終點而是業務價值的起點——有了數據你才能畫曲線、做分析、觸發告警、訓練模型。