# ESP32-S3 生产级 MQTT 通信实战指南

> 面向生产环境的 ESP32-S3 + MicroPython MQTT 通信方案。
> 完整实现：**WiFi 管理、断线重连（指数退避）、自动重订阅、应用层心跳、MQTT 协议保活、Last Will 遗嘱、离线消息缓存(store-and-forward)、看门狗防卡死、干净退出**。
> 代码全部带详细注释，可直接落地。

---

## 目录

1. [生产环境核心要点](#1-生产环境核心要点)
2. [整体架构](#2-整体架构)
3. [依赖与配置](#3-依赖与配置)
4. [核心实现：mqtt_manager.py](#4-核心实现mqtt_managerpy)
5. [整合示例：main.py](#5-整合示例mainpy)
6. [测试与验证](#6-测试与验证)
7. [生产部署最佳实践](#7-生产部署最佳实践)
8. [常见问题排查](#8-常见问题排查)

---

## 1. 生产环境核心要点

普通 Demo 的 MQTT 代码通常「连上就发」，一旦网络抖动就会**静默死机**。生产环境必须满足：

| 能力 | 说明 | 本指南实现位置 |
| --- | --- | --- |
| **断线重连** | 网络断开后自动恢复，不卡死 | `run()` 主循环 + `_connect_broker()` |
| **指数退避** | 重连间隔 1s→2s→4s…上限 60s，避免雪崩式重连打爆 Broker | `_backoff` 逻辑 |
| **自动重订阅** | 重连后必须重新订阅，否则收不到消息 | `self.subscriptions` 列表 |
| **心跳监测** | 周期性上报 `heartbeat`，应用层证明「我还活着」 | `_last_hb` + `heartbeat_interval` |
| **协议保活** | 按 `keepalive` 定时发 PING，防止 Broker 踢空闲连接 | `client.ping()` 定时触发 |
| **Last Will 遗嘱** | 设备异常掉线时 Broker 自动广播离线，后端秒级感知 | `set_last_will()` |
| **离线消息缓存** | 断线期间发出的消息先入队，恢复后补发 | `self._outbox` 队列 |
| **看门狗** | 主循环卡死时硬件复位，避免「假死」 | `machine.WDT` |
| **干净退出** | Ctrl+C / 异常时主动 `DISCONNECT` 并清遗嘱 | `stop()` |

> 关键认知：**MQTT 的 `keepalive` 是「连接层保活」**，**应用层 `heartbeat` 是「业务层健康证明」**。二者互补——前者防连接被踢，后者让后端知道设备逻辑仍在跑。

---

## 2. 整体架构

```
┌─────────────┐         Wi-Fi          ┌──────────────┐     MQTT      ┌──────────┐
│ ESP32-S3    │ ───────────────────▶  │  MQTT Broker │ ◀──────────▶ │ 后端/App │
│ (MicroPython)│                       │ (EMQX/阿里云) │               └──────────┘
└─────────────┘
   │ 内部模块
   ├─ WiFiManager   : 联网 + 掉线重连
   ├─ MQTTManager   : 连接/重连/订阅/发布
   ├─ Heartbeat     : 定时上报健康 + 响应 ping
   ├─ LWT           : 异常掉线广播离线
   ├─ Outbox Queue  : 离线消息缓存补发
   └─ WDT           : 看门狗喂狗
```

消息主题约定（以设备 `esp32s3_001` 为例）：

| 主题 | 方向 | 用途 |
| --- | --- | --- |
| `device/esp32s3_001/cmd` | 入站 | 后端下发的控制指令（`on`/`off`/`ping`/`upgrade`） |
| `device/esp32s3_001/status` | 出站 | 设备状态（retain，含 `online` 标志） |
| `device/esp32s3_001/heartbeat` | 出站 | 周期心跳（uptime / 剩余内存） |

---

## 3. 依赖与配置

### 3.1 安装 MQTT 库
标准镜像不含 MQTT，用 `mip` 安装（板子需先联网）：

```bash
mpremote connect COM3 mip install umqtt.simple
# 或板子 REPL： import mip; mip.install("umqtt.simple")
```

### 3.2 配置文件 `config.json`
**密钥与参数一律外置，禁止硬编码**。上传到板子文件系统：

```json
{
  "client_id": "esp32s3_001",
  "wifi_ssid": "YOUR_WIFI_SSID",
  "wifi_password": "YOUR_WIFI_PASSWORD",
  "broker": "broker.emqx.io",
  "port": 1883,
  "user": null,
  "password": null,
  "keepalive": 60,
  "ssl": false,
  "topic_cmd": "device/esp32s3_001/cmd",
  "topic_status": "device/esp32s3_001/status",
  "topic_heartbeat": "device/esp32s3_001/heartbeat",
  "heartbeat_interval": 30,
  "max_backoff": 60,
  "max_queue": 20,
  "wdt_timeout_ms": 8000
}
```

> 生产环境务必把 `ssl` 设为 `true`（见第 7 节），`user`/`password` 用云厂商提供的设备凭证。

---

## 4. 核心实现：mqtt_manager.py

下面是实现全部生产能力的封装类。**复制为 `mqtt_manager.py` 上传到板子即可**。

```python
# mqtt_manager.py
# ESP32-S3 生产级 MQTT 通信封装
# 功能：WiFi 管理 / 断线重连(指数退避) / 自动重订阅 / 应用层心跳 /
#       MQTT 协议保活 / Last Will 遗嘱 / 离线消息缓存(store-and-forward) / 看门狗

import network
import time
import ujson
import gc
import machine
from umqtt.simple import MQTTClient


class MQTTManager:
    """生产级 MQTT 客户端管理器（同步非阻塞实现）。"""

    def __init__(self, config: dict):
        self.cfg = config
        self.client = None                 # 当前 MQTTClient 实例
        self.connected = False             # 是否已连上 Broker
        self.subscriptions = []            # 订阅列表 [(topic, qos)...]，重连后需重订阅
        self._outbox = []                  # 离线消息队列（store-and-forward）
        self._user_cb = None               # 用户消息回调
        self._boot = time.time()           # 启动时间戳，用于计算 uptime
        self._last_ping = 0                # 上次协议层 PING 时间
        self._last_hb = 0                  # 上次应用层心跳时间
        self._backoff = 1                  # 重连退避秒数（指数增长）
        self._running = False              # 主循环运行标志
        self.wdt = self._init_wdt()        # 看门狗（失败则为 None）

    # ---------- 看门狗 ----------
    def _init_wdt(self):
        """初始化硬件看门狗。部分固件不支持，失败则降级为 None。"""
        try:
            return machine.WDT(timeout=self.cfg.get("wdt_timeout_ms", 8000))
        except Exception as e:
            print("[WDT] 初始化失败，已禁用:", e)
            return None

    # ---------- WiFi 管理 ----------
    def connect_wifi(self, max_retries: int = 20) -> bool:
        """连接 Wi-Fi，带重试。返回是否成功。"""
        wlan = network.WLAN(network.STA_IF)
        wlan.active(True)
        if wlan.isconnected():
            return True
        print("[WiFi] 连接", self.cfg["wifi_ssid"], "...")
        wlan.connect(self.cfg["wifi_ssid"], self.cfg["wifi_password"])
        # 每 0.5s 检查一次，最多等待 max_retries*0.5 秒
        for _ in range(max_retries * 2):
            if wlan.isconnected():
                print("[WiFi] 已连接, IP =", wlan.ifconfig()[0])
                return True
            time.sleep(0.5)
        print("[WiFi] 连接超时")
        return False

    def _ensure_wifi(self) -> bool:
        """确保 Wi-Fi 在线；掉线则尝试重连。"""
        wlan = network.WLAN(network.STA_IF)
        if wlan.isconnected():
            return True
        return self.connect_wifi()

    # ---------- 订阅管理 ----------
    def subscribe(self, topic: str, qos: int = 0):
        """登记并立即订阅主题（重连后自动重订阅）。"""
        pair = (topic, qos)
        if pair not in self.subscriptions:
            self.subscriptions.append(pair)
        if self.connected and self.client:
            self.client.subscribe(topic, qos)

    # ---------- 连接 Broker ----------
    def _connect_broker(self):
        """建立 MQTT 连接，配置遗嘱/回调/订阅/上线广播。失败抛异常由调用方处理。"""
        client = MQTTClient(
            client_id=self.cfg["client_id"],
            server=self.cfg["broker"],
            port=self.cfg.get("port", 1883),
            user=self.cfg.get("user"),
            password=self.cfg.get("password"),
            keepalive=self.cfg.get("keepalive", 60),
            ssl=self.cfg.get("ssl", False),
            ssl_params=self.cfg.get("ssl_params", {}),
        )

        # Last Will：设备异常掉线时，Broker 自动以 retain 发布离线状态
        client.set_last_will(
            self.cfg["topic_status"],
            ujson.dumps({"online": False, "reason": "lwt"}),
            retain=True,
            qos=1,
        )
        # 注册消息回调（在 check_msg 时被调用）
        client.set_callback(self._on_message)
        # 建立 TCP/TLS 连接并发送 CONNECT
        client.connect()

        # 重连后必须重新订阅，否则收不到消息
        for topic, qos in self.subscriptions:
            client.subscribe(topic, qos)

        # 上线广播（retain，让新订阅者立即可知设备在线）
        client.publish(
            self.cfg["topic_status"],
            ujson.dumps({"online": True}),
            retain=True,
            qos=1,
        )

        self.client = client
        self.connected = True
        self._last_ping = time.time()      # 重置保活计时
        self._backoff = 1                  # 成功连接，退避复位
        print("[MQTT] 已连接 Broker:", self.cfg["broker"])

    # ---------- 消息回调 ----------
    def _on_message(self, topic: bytes, msg: bytes):
        """Broker 推送消息时的入口。先处理心跳探针，再转交用户回调。"""
        topic_s = topic.decode()
        msg_s = msg.decode()
        print("[MQTT] 收到:", topic_s, "=>", msg_s)

        # 心跳探针：收到 "ping" 立即回 "pong"（业务层保活互证）
        if topic_s == self.cfg["topic_cmd"] and msg_s == "ping":
            self.publish(
                self.cfg["topic_status"],
                ujson.dumps({"online": True, "ts": time.time()}),
                retain=False,
                qos=0,
            )
            return

        # 转交业务回调
        if self._user_cb:
            self._user_cb(topic_s, msg_s)

    # ---------- 发布（带离线缓存） ----------
    def publish(self, topic: str, msg, retain: bool = False, qos: int = 0) -> bool:
        """发布消息。在线则直发；离线则进队列，恢复后补发。"""
        # msg 允许传 dict，自动转 JSON
        if isinstance(msg, (dict, list)):
            msg = ujson.dumps(msg)

        if self.connected and self.client:
            try:
                self.client.publish(topic, msg, retain=retain, qos=qos)
                return True
            except Exception as e:
                print("[MQTT] 发布异常，转离线队列:", e)
                self.connected = False

        # 离线：缓存（限制长度，防止内存溢出拖垮设备）
        if len(self._outbox) < self.cfg.get("max_queue", 20):
            self._outbox.append((topic, msg, retain, qos))
        else:
            print("[MQTT] 离线队列已满，丢弃消息:", topic)
        return False

    def _flush_outbox(self):
        """连接恢复后，按顺序补发离线消息。"""
        while self._outbox:
            topic, msg, retain, qos = self._outbox[0]
            try:
                self.client.publish(topic, msg, retain=retain, qos=qos)
                self._outbox.pop(0)
            except Exception:
                break  # 发送失败则保留剩余消息，下次再试

    # ---------- 主循环 ----------
    def run(self, user_cb=None):
        """阻塞式主循环：维护连接、驱动心跳与保活、处理消息。"""
        self._user_cb = user_cb
        self._running = True
        hb_interval = self.cfg.get("heartbeat_interval", 30)
        keepalive = self.cfg.get("keepalive", 60)

        try:
            while self._running:
                # 1) 喂狗（任何分支都要先喂，防卡死复位）
                if self.wdt:
                    self.wdt.feed()

                # 2) 未连接：确保 WiFi → 连接 Broker → 补发离线消息
                if not self.connected:
                    if not self._ensure_wifi():
                        time.sleep(self._backoff)
                        continue
                    try:
                        self._connect_broker()
                        self._flush_outbox()
                    except Exception as e:
                        print("[MQTT] 连接失败:", e)
                        self.connected = False
                        self.client = None
                        # 指数退避：1→2→4→...→max_backoff
                        time.sleep(self._backoff)
                        self._backoff = min(self._backoff * 2,
                                            self.cfg.get("max_backoff", 60))
                        continue

                # 3) 已连接：处理入站消息（非阻塞，并驱动 keepalive）
                try:
                    self.client.check_msg()
                except Exception as e:
                    print("[MQTT] 通信异常，标记断线:", e)
                    self.connected = False
                    self.client = None
                    continue

                # 4) 协议层保活：按 keepalive 一半周期发 PING，防被 Broker 踢
                now = time.time()
                if keepalive and (now - self._last_ping) >= keepalive // 2:
                    try:
                        self.client.ping()
                        self._last_ping = now
                    except Exception:
                        self.connected = False
                        self.client = None
                        continue

                # 5) 应用层心跳：周期上报健康指标
                if now - self._last_hb >= hb_interval:
                    self._last_hb = now
                    self.publish(
                        self.cfg["topic_heartbeat"],
                        {
                            "uptime": int(now - self._boot),
                            "free_mem": gc.mem_free(),
                            "queue": len(self._outbox),
                        },
                        retain=False,
                        qos=0,
                    )
                    gc.collect()  # 顺手回收内存

                time.sleep(0.1)  # 让出调度，避免 100% 占满 CPU

        except KeyboardInterrupt:
            print("[MQTT] 收到中断，准备退出")
        finally:
            self.stop()

    # ---------- 干净退出 ----------
    def stop(self):
        """停止循环并主动断开，清空在线状态（清理遗嘱）。"""
        self._running = False
        if self.client:
            try:
                # 主动广播离线，再 DISCONNECT
                self.client.publish(
                    self.cfg["topic_status"],
                    ujson.dumps({"online": False, "reason": "shutdown"}),
                    retain=True,
                    qos=1,
                )
                self.client.disconnect()
            except Exception:
                pass
            self.client = None
        self.connected = False
        print("[MQTT] 已断开")
```

---

## 5. 整合示例：main.py

把 MQTT 接到 LED 控制（呼应 LED 教程），并实现「后端下发指令 → 设备执行 → 上报状态」闭环：

```python
# main.py
# ESP32-S3 MQTT + LED 控制整合示例
import ujson
import machine
from mqtt_manager import MQTTManager

# 1) 加载配置（密钥不硬编码）
with open("config.json") as f:
    cfg = ujson.load(f)

# 2) 硬件：板载/外接 LED（按你的板子改引脚，外接常用 GPIO21）
led = machine.Pin(21, machine.Pin.OUT)

# 3) 创建管理器
m = MQTTManager(cfg)

# 4) 订阅控制主题（重连会自动重订阅）
m.subscribe(cfg["topic_cmd"], qos=1)

# 5) 业务消息处理
def on_message(topic, msg):
    print("命令处理:", topic, msg)
    if msg == "on":
        led.on()
    elif msg == "off":
        led.off()
    elif msg == "upgrade":
        print("[业务] 收到升级指令（调用第12节 OTA 逻辑）")
        # from ota import check_and_update; check_and_update()
    # 执行后上报最新状态（retain，新订阅者立即可知）
    m.publish(cfg["topic_status"],
              {"online": True, "led": led.value()},
              retain=True, qos=1)

# 6) 启动主循环（阻塞，内部自动处理重连/心跳/保活）
m.run(user_cb=on_message)
```

**上传与运行**：
```bash
mpremote connect COM3 fs cp config.json :config.json
mpremote connect COM3 fs cp mqtt_manager.py :mqtt_manager.py
mpremote connect COM3 fs cp main.py :main.py
mpremote connect COM3 reset
```

---

## 6. 测试与验证

用电脑侧的 **MQTTX**（https://mqttx.app）或命令行 `mosquitto_sub/pub` 验证：

```bash
# 监听设备状态与心跳
mosquitto_sub -h broker.emqx.io -t 'device/esp32s3_001/#' -v

# 下发开灯指令
mosquitto_pub -h broker.emqx.io -t 'device/esp32s3_001/cmd' -m 'on'

# 心跳探针
mosquitto_pub -h broker.emqx.io -t 'device/esp32s3_001/cmd' -m 'ping'
```

**验证清单**：
- [ ] 设备上电后自动连 Wi-Fi 并连 Broker，`status` 主题出现 `{"online":true}`（retain）
- [ ] 每 `heartbeat_interval` 秒收到一次 `heartbeat`（含 uptime / free_mem）
- [ ] 下发 `on`/`off`，LED 变化且 `status` 回显 `led` 状态
- [ ] 拔掉 Wi-Fi 再插回：设备自动重连，期间下发的指令不丢（离线队列补发）
- [ ] 突然断电重启：后端通过 LWT 立即收到 `{"online":false}`（reason: lwt）

---

## 7. 生产部署最佳实践

1. **唯一 `client_id`**：每设备用唯一 ID（如 MAC 地址），避免重复导致互相踢线。
2. **务必启用 TLS**：生产把 `ssl: true`；EMQX 公共 TLS 端口 `8883`。ESP32 需提供 CA：
   ```json
   "ssl": true,
   "ssl_params": {"ca_certs": "/flash/ca.pem"}
   ```
   将 Broker 的 CA 证书存为 `ca.pem` 上传到板子。
3. **QoS 与 retain 策略**：
   - 控制指令用 `QoS 1`（至少一次送达）；
   - 状态主题用 `retain=true`，新订阅者立即可知当前状态；
   - 高频心跳用 `QoS 0` 降开销。
4. **遗嘱 + 心跳双重检测**：LWT 秒级感知异常掉线，心跳证明业务逻辑存活。
5. **指数退避**：防网络恢复瞬间的重连风暴（见 `_backoff`）。
6. **配置外置**：`config.json` 不硬编码密钥，便于批量烧录不同设备。
7. **内存保护**：离线队列设上限（`max_queue`），主循环定期 `gc.collect()`。
8. **看门狗兜底**：`WDT` 防主循环假死，是最后一道保险。
9. **日志分级**：生产可把 `print` 换成可关闭的日志函数，降低串口开销。
10. **会话持久化**：需要离线不丢订阅/消息时，连接用 `clean_session=False`（umqtt 在 `connect(clean_session=False)` 指定），并依赖 Broker 的持久会话。

---

## 8. 常见问题排查

| 现象 | 原因 / 解决 |
| --- | --- |
| 频繁掉线又重连 | `client_id` 重复，或 `keepalive` 太小且未调用 `check_msg` → 用唯一 ID，保证主循环在跑 |
| 重连后收不到消息 | 未重订阅 → 本封装已用 `subscriptions` 自动重订阅，确保用 `subscribe()` 登记 |
| 心跳正常但 `status` 不更新 | 发布时连接已被标记断线，消息进了离线队列但 Broker 不可达 → 检查 WiFi |
| LWT 未触发离线 | 仅异常断开才触发；正常 `stop()` 会主动发离线 → 断电/看门狗复位才会走 LWT |
| 内存逐渐耗尽 | 离线队列无限增长 / 未 `gc.collect()` → 本封装已限长 + 定期回收 |
| TLS 连接失败 | 缺 CA 证书或 `ssl_params` 配置错误 → 上传正确 `ca.pem` 并指向路径 |
| `import umqtt` 报错 | 未安装库 → 用 `mip install umqtt.simple` |

---

> 配套教程《ESP32-S3 MicroPython 全栈实战指南》中第 9~12 节的 Wi-Fi、Web 控制、OTA 升级与本文档的 MQTT 可组合使用：用 MQTT 触发 `upgrade` 即可实现**手机远程一键 OTA**。