ESP32 + MQTT 物联网实战：从传感器上报到 Spring Boot 订阅落库

分类：嵌入式与 IoT · 作者：管理员 · 发布于 2026-10-20 · 阅读 0 · 标签：ESP32 / MQTT / 物联网 / Spring Boot / 嵌入式

痛点：一条常见却总在半夜报警的 IoT 上报链路

做物联网项目时，设备端往往很简单：ESP32 采集温湿度、定时上报，后端订阅主题落库。但真正上线后，问题全藏在看不见的地方：设备掉电了后台却显示「在线」、网络抖动后设备不再重连、QoS 1 消息重复导致数据库出现重复记录、Broker 重启后后端再也收不到数据。本文以一个典型的「环境监测」场景为例，给出从 ESP32 设备端到 Spring Boot 服务端的完整落地方案，代码均可直接运行。

总体架构与主题设计

整体链路为：ESP32（发布端）→ MQTT Broker（消息代理）→ Spring Boot（订阅端落库）。Broker 可选 EMQX 或 Mosquitto，本文不依赖 Broker 私有特性，标准 MQTT 协议即可。

主题（Topic）设计是 IoT 系统的地基，建议采用「层级 + 设备标识」的结构：

sensor/{deviceId}/data：传感器数据上报，负载为 JSON，QoS 1；
sensor/{deviceId}/status：设备上下线状态，配合遗嘱消息（LWT）使用；
sensor/{deviceId}/cmd：服务端下发指令（如修改上报频率）。

主题要按设备隔离，避免所有设备共用同一主题后只能靠 payload 里的 deviceId 区分，否则无法按设备做权限与监控的隔离。

设备端：ESP32 的稳定连接是第一步

Arduino 框架下最常用的是 PubSubClient 库，但它只负责 MQTT 协议，Wi-Fi 断线需要自己处理。设备端最容易犯的错误是「Wi-Fi 断了只重连 Wi-Fi，MQTT 连接却一直挂在断开的 socket 上」。正确做法是：连接状态统一收敛到 MQTT 层，只要未连上 Broker 就循环重试。

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

const char* WIFI_SSID = "your-ssid";
const char* WIFI_PASS = "your-pass";
const char* MQTT_HOST = "mqtt.example.com";
const int   MQTT_PORT = 1883;

// 用芯片 MAC 尾部生成稳定且唯一的 Client ID，便于运维识别设备
String clientId = "ESP32_" + String((uint32_t)ESP.getEfuseMac(), HEX);

WiFiClient espClient;
PubSubClient mqtt(espClient);

void connectMqtt() {
  while (!mqtt.connected()) {
    Serial.printf("尝试连接 Broker，ClientId=%s\n", clientId.c_str());
    // 遗嘱消息：异常断开时 Broker 代发，通知后端该设备离线
    String will = "{\"deviceId\":\"" + clientId + "\",\"status\":\"offline\"}";
    if (mqtt.connect(clientId.c_str(), "iot-user", "iot-pass",
                     "sensor/" + clientId + "/status", 1, true, will.c_str())) {
      Serial.println("MQTT 已连接");
      // 上线主动广播一次在线状态，retain=true 让后端新订阅者即刻拿到当前状态
      mqtt.publish("sensor/" + clientId + "/status",
                   "{\"deviceId\":\"" + clientId + "\",\"status\":\"online\"}", true);
    } else {
      Serial.print("连接失败，错误码：");
      Serial.println(mqtt.state());
      delay(3000); // 失败至少等 3 秒，避免重连风暴
    }
  }
}

void setup() {
  Serial.begin(115200);
  WiFi.begin(WIFI_SSID, WIFI_PASS);
  while (WiFi.status() != WL_CONNECTED) { delay(500); }
  mqtt.setServer(MQTT_HOST, MQTT_PORT);
  mqtt.setKeepAlive(30); // 心跳 30s，Broker 约 45s（1.5 倍）后可判定掉线
}

void loop() {
  if (!mqtt.connected()) {
    connectMqtt();          // Wi-Fi 未连时 PUBLISH 会失败，连接状态统一在此收敛
  }
  mqtt.loop();              // 必须周期调用以处理心跳与收包
  static unsigned long last = 0;
  if (millis() - last > 10000) {
    last = millis();
    char payload[96];
    // ts 用毫秒时间戳转秒，保证同一设备同一秒的上报键唯一（生产环境建议先 NTP 校时再用真实时间）
    snprintf(payload, sizeof(payload),
             "{\"deviceId\":\"%s\",\"ts\":%lu,\"temp\":%.1f,\"humidity\":%.1f}",
             clientId.c_str(), millis() / 1000, 25.6, 60.3);
    // QoS 1：Broker 确认（PUBACK）后才算成功，可容忍重复但不可丢失
    mqtt.publish("sensor/" + clientId + "/data", payload, 1);
  }
}

三个关键细节：keepAlive 不要设太大（设备掉线检测延迟约为 1.5 倍心跳间隔）；重连间隔必须加退避，否则几十台设备同时断网恢复后会瞬间打爆 Broker；正常主动下线不会触发遗嘱消息，需要主动发布一次 offline 状态。

服务端：Spring Boot 订阅落库与幂等去重

服务端推荐用 HiveMQ MQTT Client（纯 Java、异步、自带自动重连），比老旧的 Paho 回调风格更简洁。依赖如下：

<dependency>
    <groupId>com.hivemq</groupId>
    <artifactId>hivemq-mqtt-client</artifactId>
    <version>1.4.1</version>
</dependency>

@Configuration
public class MqttConfig {

    @Bean(destroyMethod = "disconnect")
    public Mqtt5AsyncClient mqttClient(DeviceDataService service) {
        Mqtt5AsyncClient client = Mqtt5Client.builder()
                .identifier("iot-backend")
                .serverHost("mqtt.example.com")
                .serverPort(1883)
                .automaticReconnectWithDefaultConfig() // 断线自动重连 + 指数退避
                .addConnectedListener(context -> {      // 每次（重）连接成功后重新订阅
                    context.getClient().subscribeWith()
                            .topicFilter("sensor/+/data")
                            .qos(MqttQos.AT_LEAST_ONCE)
                            .callback(msg -> service.handle(msg.getTopic(), msg.getPayloadAsBytes()))
                            .send();
                })
                .buildAsync();
        client.connect();
        return client;
    }
}

@Service
@Slf4j
public class DeviceDataService {

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final JdbcTemplate jdbcTemplate;

    public DeviceDataService(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    public void handle(String topic, byte[] payload) {
        SensorData data;
        try {
            data = objectMapper.readValue(payload, SensorData.class);
        } catch (Exception e) {
            log.warn("非法上报数据，topic={}", topic); // 脏数据直接丢弃并告警，不要阻塞消费
            return;
        }
        // 幂等写入：同一设备同一秒的上报只落库一次（依赖 PostgreSQL 唯一索引）
        int updated = jdbcTemplate.update(
            "INSERT INTO sensor_data(device_id, ts, temp, humidity) VALUES (?, ?, ?, ?) " +
            "ON CONFLICT (device_id, ts) DO NOTHING",
            data.deviceId(), data.ts(), data.temp(), data.humidity());
        if (updated == 0) {
            log.info("重复上报已去重：deviceId={}, ts={}", data.deviceId(), data.ts());
        }
    }

    public record SensorData(String deviceId, long ts, double temp, double humidity) {}
}

这里做了两件容易被忽略的事：一是「连接成功」监听里订阅，而不是启动时订阅一次——Broker 重启或网络分区后重连成功会自动恢复订阅；二是用 (device_id, ts) 唯一索引 + ON CONFLICT DO NOTHING 做幂等，把 QoS 1 带来的重复消息挡在数据库层，比先查后插更可靠、也更快。

踩坑记录

坑 1：设备显示「在线」却是僵尸连接。只依赖 TCP 层判断。解决：Broker 侧开 Keep Alive 踢除 + 遗嘱消息兜底，后端以 LWT 为准更新设备状态。
坑 2：QoS 1 消息重复导致报表翻倍。QoS 1 语义是「至少一次」，网络重传、客户端重连都会造成重复。解决：落库幂等（唯一索引），而不是在消费端 try-catch 里做内存去重——应用重启后内存去重会失效。
坑 3：后端重启后收不到数据。默认 clean session 时，Broker 不会为离线订阅者缓存消息。解决：按本文方式在连接监听中重新订阅；若要求不丢消息，可开持久会话并把 QoS 提到 1，代价是 Broker 内存占用上升。
坑 4：JSON 字段类型不匹配。ESP32 端 snprintf 输出的小数可能被网关转成字符串，后端用 Double 解析抛异常。解决：设备端固定格式并做类型校验，解析失败进死信而不是静默丢弃。
坑 5：重连风暴。断电恢复后所有设备同时重连。解决：设备端随机抖动退避（如 3s + random(0,5)s），Broker 端限制连接速率。

最佳实践清单

主题按「层级 + 设备 ID」设计，数据、状态、指令分开，便于 ACL 与监控；
设备端 keepAlive 建议 10~60s，掉线感知与流量开销取平衡；
遗嘱消息 + 保留消息配合使用，但不要把它当作唯一在线检测手段，它只对异常断开生效且有 1.5 倍心跳延迟；
服务端订阅放在连接成功回调中，天然覆盖重连场景；
数据库唯一索引做幂等，比应用层去重更可靠；
解析失败的消息要进死信队列并告警，禁止静默丢弃，否则数据缺口无从排查。

这套链路同样适用于温控、门锁、环境监测等场景：把「设备稳定上报、后端可靠落库」的地基打好，上层业务（告警、报表、联动）才不会在半夜给你打电话。