ESP32 + MQTT + Spring Boot 物联网数据采集实战：从传感器到服务端持久化

分类：嵌入式与 IoT · 作者：管理员 · 发布于 2026-09-27 · 阅读 0 · 标签：ESP32 / MQTT / Spring Boot / IoT / 传感器

一、场景痛点

物联网（IoT）项目中最常见的需求就是：设备端采集传感器数据 → 通过网络传输 → 服务端接收、存储、处理。HTTP 轮询模式在设备量少时尚可接受，一旦设备数量增长到几十上百台，HTTP 的短连接开销、服务端被动等待、设备端高功耗轮询等缺陷就暴露无遗。

MQTT（Message Queuing Telemetry Transport）协议专为 IoT 场景设计——轻量级、发布/订阅模式、支持 QoS 等级、极低带宽消耗，天然适合 ESP32 这类资源受限设备与服务端之间的双向通信。

本文以一个真实场景为例：ESP32 采集温湿度传感器数据（DHT22 / SHT30），通过 MQTT 上报，Spring Boot 服务端订阅并持久化到数据库，完整串联设备端 → 消息中间件 → 业务后端的全链路。

二、整体架构

┌─────────────────┐     MQTT Pub      ┌──────────────┐     MQTT Sub      ┌──────────────────────┐
│  ESP32 (Client) │ ────────────────→ │  MQTT Broker  │ ────────────────→ │  Spring Boot (Sub)   │
│  (DHT22 传感器)  │    topic:         │  (Mosquitto)  │    topic:         │  ├─ MQTT 回调处理     │
│                 │   sensor/temp     │               │   sensor/temp    │  ├─ 数据校验与转换     │
│                 │   QoS 1           │               │   QoS 1          │  └─ JPA 持久化 → DB  │
└─────────────────┘                   └──────────────┘                   └──────────────────────┘

设备端：ESP32 运行 Arduino 框架，连接 Wi-Fi，周期采集传感器数据，通过 MQTT 客户端发布到 Broker。
消息代理：Mosquitto（轻量开源 MQTT Broker，可 Docker 部署），负责消息路由与 QoS 保障。
服务端：Spring Boot 应用作为 MQTT 订阅者，监听指定 topic，收到消息后解析 JSON、校验、持久化到 MySQL/PostgreSQL。

三、ESP32 端：传感器采集与 MQTT 上报

使用 Arduino IDE 或 PlatformIO，引入 PubSubClient 库和 DHT 传感器库。

#include <WiFi.h>
#include <PubSubClient.h>
#include <DHT.h>

#define DHTPIN 4
#define DHTTYPE DHT22
DHT dht(DHTPIN, DHTTYPE);

const char* WIFI_SSID = "YourWiFiSSID";
const char* WIFI_PASS = "YourWiFiPassword";
const char* MQTT_BROKER = "mqtt.example.com";
const int   MQTT_PORT = 1883;
const char* MQTT_TOPIC = "sensor/temp";
const char* MQTT_CLIENT_ID = "esp32_sensor_01";

WiFiClient wifiClient;
PubSubClient mqttClient(wifiClient);

void setupWiFi() {
  WiFi.begin(WIFI_SSID, WIFI_PASS);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println("\nWiFi connected");
}

void reconnectMQTT() {
  while (!mqttClient.connected()) {
    if (mqttClient.connect(MQTT_CLIENT_ID)) {
      Serial.println("MQTT connected");
    } else {
      Serial.print("MQTT connect fail, rc=");
      Serial.println(mqttClient.state());
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200);
  dht.begin();
  setupWiFi();
  mqttClient.setServer(MQTT_BROKER, MQTT_PORT);
}

void loop() {
  if (!mqttClient.connected()) {
    reconnectMQTT();
  }
  mqttClient.loop();

  float h = dht.readHumidity();
  float t = dht.readTemperature();

  if (isnan(h) || isnan(t)) {
    Serial.println("Sensor read failed");
    delay(5000);
    return;
  }

  char payload[128];
  snprintf(payload, sizeof(payload),
    "{\"deviceId\":\"esp32_01\",\"temperature\":%.1f,\"humidity\":%.1f,\"timestamp\":%lu}",
    t, h, millis() / 1000);

  boolean published = mqttClient.publish(MQTT_TOPIC, payload, true);
  if (published) {
    Serial.print("Published: ");
    Serial.println(payload);
  } else {
    Serial.println("Publish failed");
  }

  delay(10000); // 每 10 秒上报一次
}

关键点说明：

publish(..., true) 第三个参数 retained=true，Broker 会保留最后一条消息，新订阅者上线即可获取最新值，避免空窗期。
QoS 默认 0（至多一次），需要可靠性时改为 publish(topic, payload, false) 后在 PubSubClient 构造函数中设置 mqttClient.setQoS(1)。实测建议用 QoS 1 保证至少一次送达。
断线重连逻辑必不可少——ESP32 和 Broker 之间 TCP 连接可能因 Wi-Fi 不稳定断开，reconnectMQTT() 每轮 loop 检查并自动恢复。

四、MQTT Broker 部署：Mosquitto Docker 版

Mosquitto 是最轻量的开源 MQTT Broker 实现，官方 Docker 镜像一行命令启动：

docker run -d --name mqtt-broker \
  -p 1883:1883 \
  -p 9001:9001 \
  -v /opt/app/mosquitto/config:/mosquitto/config \
  -v /opt/app/mosquitto/data:/mosquitto/data \
  eclipse-mosquitto:2

配置 mosquitto.conf：

listener 1883
allow_anonymous true

# 如果启用认证（推荐生产环境）
# password_file /mosquitto/config/passwd
# allow_anonymous false

listener 9001
protocol websockets

1883：标准 MQTT 端口（TCP）
9001：WebSocket 端口，用于浏览器或 Node.js 调试客户端
生产环境务必配置 password_file 开启用户名/密码认证

五、Spring Boot 服务端：MQTT 订阅与数据处理

引入 Eclipse Paho MQTT 客户端依赖（Gradle 示例）：

implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'

编写 MQTT 配置与回调类，核心代码分三部分：

5.1 MQTT 配置类

@Configuration
public class MqttConfig {

    @Value("${mqtt.broker:tcp://mqtt.example.com:1883}")
    private String brokerUrl;

    @Value("${mqtt.client-id:backend-subscriber}")
    private String clientId;

    @Value("${mqtt.topic:sensor/#}")
    private String topic;

    @Bean
    public MqttClient mqttClient() throws MqttException {
        MqttClient client = new MqttClient(brokerUrl, clientId);
        MqttConnectOptions opts = new MqttConnectOptions();
        opts.setCleanSession(false);           // 断线恢复时接收离线消息
        opts.setAutomaticReconnect(true);       // 自动重连
        opts.setConnectionTimeout(10);
        opts.setKeepAliveInterval(30);          // 心跳间隔（秒）
        opts.setMaxReconnectDelay(60000);       // 最大重连间隔
        // 若启用认证
        // opts.setUserName("admin");
        // opts.setPassword("******".toCharArray());

        client.setCallback(new SensorDataCallback());
        client.connect(opts);
        client.subscribe(topic, 1);  // QoS 1：至少一次
        return client;
    }
}

5.2 消息回调处理

@Component
public class SensorDataCallback implements MqttCallback {

    @Autowired
    private SensorDataService sensorDataService;

    @Override
    public void connectionLost(Throwable cause) {
        // Paho 设置了自动重连，此处只记录日志
        log.warn("MQTT connection lost: {}", cause.getMessage());
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 仅发布者（Publish）场景需要
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
        log.info("Received [{}]: {}", topic, payload);

        try {
            SensorData data = JSON.parseObject(payload, SensorData.class);
            // 时间戳校验：拒绝超过 5 分钟的过期数据
            long now = System.currentTimeMillis() / 1000;
            if (now - data.getTimestamp() > 300) {
                log.warn("Stale data discarded: {}", data.getDeviceId());
                return;
            }
            sensorDataService.save(data);
        } catch (Exception e) {
            log.error("Parse message failed: {}", e.getMessage());
        }
    }
}

5.3 数据实体与持久化

@Entity
@Table(name = "sensor_data")
public class SensorData {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(nullable = false, length = 64)
    private String deviceId;

    @Column(nullable = false)
    private Double temperature;

    @Column(nullable = false)
    private Double humidity;

    @Column(nullable = false)
    private Long timestamp;

    @Column(nullable = false, updatable = false)
    private LocalDateTime createTime;

    @PrePersist
    public void prePersist() {
        this.createTime = LocalDateTime.now();
    }
}

六、踩坑记录

6.1 QoS 选择误区

一开始为了"可靠性"全部用 QoS 2（恰好一次），结果 ESP32 内存溢出——QoS 2 的 4 次握手协议报文对 ESP32 的 320KB RAM 是巨大压力。实测结论：传感器数据用 QoS 1 足够，QoS 2 仅用于控制指令（如远程开关、固件升级指令）。

6.2 心跳保活与 Paho 自动重连

Spring Boot 端使用 Paho 客户端时，setAutomaticReconnect(true) 配合 setMaxReconnectDelay(60000) 能优雅处理 Broker 短暂重启。但注意：Broker 重启后，Paho 自动重连成功后需重新 subscribe topic。解决方法是在 connectionLost() 中设置标记，重连后在 connectComplete() 回调中重新 subscribe。

6.3 消息积压与消费能力不匹配

如果 ESP32 设备数量多（几十台 +），每 10 秒一条消息，服务端处理不过来就会造成 MQTT 消息在 Broker 侧堆积。解决方案：

服务端引入线程池异步处理——@Async + 自定义 ThreadPoolTaskExecutor
写入数据库前用批量插入（jdbc.batchSize=100）替代逐条 INSERT
对历史趋势数据启用 时序数据库（如 TimescaleDB 或 InfluxDB）替代关系库

6.4 重复消息与消费幂等

QoS 1 在极端情况下（ACK 丢包）可能产生重复消息。数据库层用 deviceId + timestamp 做唯一索引或先查后插来去重：

ALTER TABLE sensor_data ADD UNIQUE INDEX uk_device_time (device_id, timestamp);

七、最佳实践总结

topic 命名规范：使用分层结构如 sensor/{deviceId}/temp、sensor/{deviceId}/humidity，服务端用通配符 sensor/+/temp 订阅，后续扩展设备无需改代码。
payload 标准化：统一 JSON 格式 + 时间戳字段，服务端按时间戳做过期过滤（拒绝 5 分钟前的数据）。
离线缓存：关键设备开启 cleanSession=false，Broker 缓存 QoS 1/2 消息，设备重连后自动补发。
监控告警：服务端统计每分钟收到的消息量，连续 3 分钟为 0 则触发告警——可能设备集体掉线或 Broker 挂了。
安全底线：MQTT 不加密时 payload 和密码均为明文，生产环境必须上 TLS（端口 8883）或 VPN 隧道。

以上就是 ESP32 + MQTT + Spring Boot 物联网数据采集的完整实战链路。从硬件采集到服务端持久化，每一步都有值得注意的细节——QoS 取舍、断线重连、消息积压、幂等消费。希望这套模式能帮你快速搭建自己的 IoT 数据平台。