ESP32-S3 跑 MicroPython 连 EMQX

ESP32-S3 跑 MicroPython 连 EMQX，本质上是用 umqtt.simple 这个库先连 Wi-Fi、再发起 MQTT 连接。下面给你一套能直接跑的代码，并说明几个容易踩的坑。

一、最简可运行代码

先把 umqtt.simple 放到 ESP32-S3 的 /lib 下（如果你的固件没自带，用 upip 安装或手动上传 umqtt.py），然后写 main.py：

import network
import time
import ubinascii
from umqtt.simple import MQTTClient

# ---------- 1. Wi-Fi 配置 ----------
WIFI_SSID = "你的WiFi名"
WIFI_PASS = "你的WiFi密码"

# ---------- 2. EMQX 配置 ----------
# 公共服务器（测试用）
MQTT_SERVER = "broker.emqx.io"
MQTT_PORT = 1883               # TCP 明文
# MQTT_PORT = 8883             # TLS 加密（Serverless 部署强制要求）

MQTT_CLIENT_ID = ubinascii.hexlify(ubinascii.hexlify(network.WLAN().config('mac'))).decode()
MQTT_USER = "emqx"             # 若 EMQX 开启认证则填你的用户名，否则留空串 ""
MQTT_PASS = "public"           # 同上
PUB_TOPIC = "esp32s3/pub"
SUB_TOPIC = b"esp32s3/sub"

# ---------- 3. 连 Wi-Fi ----------
def connect_wifi():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print("Connecting to WiFi...")
        wlan.connect(WIFI_SSID, WIFI_PASS)
        while not wlan.isconnected():
            time.sleep(0.5)
    print("WiFi connected:", wlan.ifconfig()[0])

# ---------- 4. MQTT 回调 ----------
def on_message(topic, msg):
    print(f"Received: {topic} -> {msg}")

# ---------- 5. 主流程 ----------
def main():
    connect_wifi()

    client = MQTTClient(
        MQTT_CLIENT_ID, MQTT_SERVER, port=MQTT_PORT,
        user=MQTT_USER, password=MQTT_PASS
    )
    client.set_callback(on_message)
    client.connect()
    print(f"Connected to EMQX at {MQTT_SERVER}:{MQTT_PORT}")

    client.subscribe(SUB_TOPIC)
    print(f"Subscribed: {SUB_TOPIC}")

    while True:
        client.check_msg()                    # 非阻塞检查消息
        client.publish(PUB_TOPIC, "hello from ESP32-S3")
        time.sleep(5)

if __name__ == "__main__":
    main()

二、EMQX 连接信息怎么填

EMQX 的端口和协议对应关系固定，按你的部署类型选 ：

⚠️ 三个高频踩坑点：

1. Client ID 必须唯一 ：多台设备用同一个 Client ID 会互相踢下线。上面代码用 MAC 地址生成，基本不会撞。
2. EMQX Serverless 只开放 8883 TLS 端口 ：如果你用的是 EMQX Cloud Serverless，必须把 MQTT_PORT 改成 8883，且 MQTTClient 要带 ssl=True 并指定 CA 证书。
3. 开了认证就必须填用户名密码 ：在 EMQX Dashboard 的 访问控制 > 客户端认证 里添加账号；如果服务端是匿名认证，用户名密码传空串 "" 即可。

三、TLS 连接（8883 端口）怎么写

MicroPython 的 SSL 支持比较简陋，连 EMQX Cloud 的 8883 时必须指定 CA 证书，否则握手失败 ：

import ussl

# 把 EMQX 提供的 CA 证书存为 /ca.crt，然后在连接前：
with open("/ca.crt", "rb") as f:
    ca_cert = f.read()

client = MQTTClient(
    MQTT_CLIENT_ID, MQTT_SERVER, port=8883,
    user=MQTT_USER, password=MQTT_PASS,
    ssl=True,
    ssl_params={"cert": ca_cert}     # 校验服务器证书
)
client.connect()

如果只是为了测试、想跳过证书校验，可以用 ssl_params={"verify_mode": ussl.CERT_NONE}，但生产环境千万别这么干。

四、断线重连与保活

umqtt.simple 只支持 MQTT 3.1.1，不支持 QoS2 ，而且默认不带自动重连。生产代码建议套一层 umqtt.robust：

from umqtt.robust import MQTTClient   # 自带断线重连

client = MQTTClient(MQTT_CLIENT_ID, MQTT_SERVER, port=MQTT_PORT,
                    user=MQTT_USER, password=MQTT_PASS, keepalive=60)

主循环里用 client.check_msg() 做非阻塞接收，掉线后 umqtt.robust 会在下次 check_msg() / publish() 时自动重连。

五、如果你的 ESP32-S3 是外挂 ML307A/ML307R 4G 模组

这种情况下不要让 ESP32 自己跑 MQTT 协议，而是让 ESP32 通过 UART 给 4G 模组发 AT 指令，由模组自己建立 MQTT 连接到 EMQX。典型流程 ：

AT+CPIN?                          # 检查 SIM 卡
AT+CEREG?                         # 检查 4G 注册
AT+MQTTCFG="version",4            # 配置 MQTT 参数
AT+MQTTCONN=0,"broker.emqx.io",1883,"ML307A","admin","public"
AT+MQTTSUB=0,"testML",1
AT+MQTTPUB=0,"testML",0,0,0,18,"{"msg":"Hello"}"

ESP32-S3 这边只需要往串口写 AT 指令、解析 +MQTTURC 类响应即可，不需要 umqtt 库。ML307A 的 AT+MQTTCONN 完整语法是 AT+MQTTCONN=<link_id>,"<host>",<port>,"<client_id>","<username>","<password>" 。

调试建议：先用电脑上的 MQTTX 客户端（填同样的 broker 地址、用户名密码）连一下 EMQX，确认服务端没问题、账号有权限，再跑 ESP32 代码，排错效率会高很多 。

你现在是直连 Wi-Fi 还是走 4G 模组？如果是 4G 模组，告诉我具体型号（ML307A 还是 ML307R），我可以把 AT 指令时序和 ESP32 串口解析代码完整写出来。