📚 课程目录第3章农业物联网数据采集与存储接口

3. 农业物联网数据采集与存储接口

在农业物联网系统中，传感器节点（土壤温湿度、空气温湿度、光照、CO₂、pH 等）产生的数据必须先“进得来”，再“存得住”。本节聚焦一条完整的数据接入链路：设备通过 MQTT 或 HTTP 把数据送到接收服务，接收服务用 Python 或 Java 解析、校验后写入 MySQL 或 PostgreSQL。理解这条链路的接口形态与落库细节，是后续查询、分析与可视化的一切前提。

一、两种接入方式的定位差异

MQTT 是发布/订阅模型，适合低带宽、长连接、高频上报的田间传感器：设备向主题（如 farm/plot01/sensor/soil）发布消息，服务端订阅该主题即可持续接收。HTTP 是请求/响应模型，适合网关定时批量上报、移动端或第三方平台主动推送，接口直观、易调试，但每次请求都有连接开销。实际工程中常见“设备走 MQTT、网关聚合后走 HTTP”的混合架构。

选择时看三个参数：上报频率、网络稳定性、设备能力。每分钟一次以上的高频采集优先 MQTT；每天几次的批量同步用 HTTP 更简单；弱网环境 MQTT 的 QoS 1 能提供至少一次投递保障，但会带来重复消息，落库时必须做幂等处理。

二、数据报文与表结构设计

无论走哪种协议，接收服务最终要处理的是同一种结构化数据。一个典型的传感器上报报文如下（示例数据，非实测）：

{
  "msg_id": "a1b2c3d4-0001",
  "device_id": "SEN-SOIL-001",
  "plot_id": "plot01",
  "ts": "2024-05-20T08:30:00Z",
  "metrics": {
    "soil_temp": 21.6,
    "soil_moisture": 38.2,
    "air_humidity": 65.0
  }
}

对应的 PostgreSQL 表结构建议采用“窄表”设计，一行一个指标，便于后续按指标聚合查询：

CREATE TABLE telemetry (
  id          BIGSERIAL PRIMARY KEY,
  msg_id      UUID        NOT NULL,
  device_id   VARCHAR(32) NOT NULL,
  plot_id     VARCHAR(32) NOT NULL,
  ts          TIMESTAMPTZ NOT NULL,
  metric      VARCHAR(32) NOT NULL,
  value       NUMERIC(10,2) NOT NULL,
  quality     SMALLINT    DEFAULT 0,
  created_at  TIMESTAMPTZ DEFAULT now(),
  UNIQUE (device_id, ts, metric)
);
CREATE INDEX idx_telemetry_ts ON telemetry (ts);

唯一索引 (device_id, ts, metric) 是幂等的核心：MQTT QoS 1 可能重复投递，HTTP 客户端也可能重试，插入时用 ON CONFLICT DO NOTHING（PostgreSQL）或 INSERT IGNORE（MySQL）即可安全去重，不会产生重复行。

三、Python 接收服务实现要点

以 Python 为例，用 paho-mqtt 订阅主题、psycopg2 写库。核心是把“接收”和“落库”解耦：回调函数只负责把消息放入内存队列，另起线程批量消费，避免数据库慢导致 MQTT 心跳超时。

import json, queue, threading
import paho.mqtt.client as mqtt
import psycopg2

q = queue.Queue(maxsize=10000)

def on_message(client, userdata, msg):
    try:
        payload = json.loads(msg.payload)
        q.put(payload, timeout=1)
    except Exception as e:
        print("解析失败:", e)

def writer():
    conn = psycopg2.connect("dbname=farm user=iot password=*** host=127.0.0.1")
    cur = conn.cursor()
    batch = []
    while True:
        item = q.get()
        for metric, value in item["metrics"].items():
            batch.append((item["msg_id"], item["device_id"],
                          item["plot_id"], item["ts"], metric, value))
        if len(batch) >= 500:
            cur.executemany(
                "INSERT INTO telemetry(msg_id,device_id,plot_id,ts,metric,value) "
                "VALUES (%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING", batch)
            conn.commit()
            batch.clear()

threading.Thread(target=writer, daemon=True).start()
client = mqtt.Client()
client.on_message = on_message
client.connect("broker.local", 1883, 60)
client.subscribe("farm/+/sensor/#", qos=1)
client.loop_forever()

Java 侧可用 Spring Boot + Eclipse Paho + JdbcTemplate 实现同样结构，关键点一致：异步队列、批量提交、冲突忽略。HTTP 接入则用 Flask/FastAPI 或 Spring MVC 暴露 POST /api/v1/telemetry，请求体校验通过后走同一个写入队列，保证两条通道共用一套落库逻辑。

四、验证与常见问题判断

服务跑起来后，用 mosquitto_pub 发一条测试消息，再查库确认：

mosquitto_pub -h broker.local -t farm/plot01/sensor/soil \
  -m '{"msg_id":"test-001","device_id":"SEN-SOIL-001","plot_id":"plot01","ts":"2024-05-20T08:30:00Z","metrics":{"soil_temp":21.6}}'

SELECT device_id, ts, metric, value FROM telemetry
WHERE device_id='SEN-SOIL-001' ORDER BY ts DESC LIMIT 5;

若查不到数据，按链路逐段排查：Broker 是否收到消息（看订阅端日志）、JSON 是否解析成功、数据库连接是否正常、唯一索引是否把重复消息全部挡掉（这属于正常现象而非故障）。若数据量突增导致写入延迟，优先调大批量阈值或增加消费者线程，而不是盲目加索引。落库成功后，created_at 与 ts 的差值还能反映端到端延迟，是判断链路健康度的实用指标。