树莓派Pico W + MicroPython + MQTT + EMQX 环境监测实战

手里这块树莓派 Pico 已经吃灰很久了，最近把它翻出来做了个环境监测节点，目标很简单：用 MicroPython 采集温湿度，通过 MQTT 协议把 JSON 消息发布到 EMQX Broker，再在电脑上实时看到数据。整个过程走下来，比我想象中要多绕好几个弯：普通 Pico 根本没有 Wi-Fi、MicroPython 新版固件不自带 MQTT 库、EMQX 装好之后默认配置还有一堆安全细节要处理……

这篇记录不是抄官方文档的流水账，而是把我从翻出板子到最终稳定收到 JSON 数据全过程中踩过的坑、验证过的方案，以及最终跑通的完整代码一起整理出来。适合手里有 Pico / Pico W、想做物联网数据上报、又不想一上来就啃 C 语言的读者。

1. 选型逻辑：为什么偏偏是 Pico + MicroPython + EMQX

1.1 为什么必须介意 Pico 与 Pico W 的区别

这是整套项目里第一个会被忽略、却最致命的点。树莓派 Pico 用的是 RP2040 芯片，本身不带任何无线模块；而 Pico W 在同样一颗 RP2040 旁边多塞了一个 CYW43439 Wi-Fi 芯片。MQTT 走的是 TCP/IP 网络，没有 Wi-Fi 模块的普通 Pico 根本没法直接上 MQTT。

我不止一次看到新手拿着普通 Pico 折腾半天，最后在network.WLAN()那行代码上报AttributeError。如果你还没买板子，直接选 Pico W；如果手里只有普通 Pico，也不是完全没救——可以外接 ESP-01S 之类模块，通过串口 AT 指令做网络透传，但那会让代码复杂一个量级，适合当成独立项目来搞，不适合作为入门 MQTT 的第一站。

1.2 MicroPython 把开发速度提到什么程度

用 C 语言 + SDK 做这件事，光连接 Wi-Fi 的 SDK 初始化代码就是几十行，还要自己管理事件回调、内存池，对只想验证业务逻辑的人来说负担太重。MicroPython 把底层封装成了现成模块，Wi-Fi 连接几行搞定，MQTT 消息直接拿字符串拼 JSON，调试的时候甚至能在 REPL 里手动敲代码看现象。

代价是内存和性能。RP2040 只有 264KB RAM，MicroPython 解释器本身要占一部分，跑大模型、高并发消息处理不现实。但作为传感器节点，每 5 秒发布一条 JSON 消息，这个负载对 MicroPython 来说完全没压力。我自己习惯把 MicroPython 定位成"快速原型验证层"，功能跑通之后再决定要不要用 C 重写。

1.3 EMQX 和 JSON 在这条链路里各管哪一段

很多人会把"MQTT 协议"和"MQTT Broker"混在一起，实际上它们是两回事。MQTT 是应用层协议，规定了客户端如何连接、订阅、发布消息；而 EMQX 是这个协议的服务端实现，负责接收所有客户端的连接，并按 Topic 把消息路由给订阅者。没有 Broker，发布者和订阅者之间就只是点对点的孤儿。

JSON 则纯粹是"消息内容格式"。MQTT 报文里承载的是一串 bytes，至于这串 bytes 是人类能读的文本、还是加密二进制、还是 ProtoBuf，协议本身不管。物联网场景里 JSON 是最通用的选择，因为上下游语言都有现成解析库，调试时直接用肉眼看也直观。我用一条 JSON 把所有信息打包：设备 ID、温度、湿度、时间戳，一次发布全带上，避免为了每个字段开好几个 Topic。

2. MQTT 机制拆解：从 CONNECT 到 PUBLISH 的消息旅程

2.1 公告板模型与 Topic 通配符

理解 MQTT 最简单的方式是把它想成一套公告板系统：Broker 是公告板本身，Topic 是公告板上的格子，每个客户端既可以往格子上贴纸条，也可以盯着某个格子等新纸条。发纸条的人和看纸条的人互相不认识，完全通过 Broker 中转。

Topic 是 MQTT 的寻址核心，我用的是层级结构，比如pico/sensor，斜杠把主题分成层级。订阅端可以用通配符批量监听：+匹配某一层的任意名字，#匹配剩余所有层级。pico/+/sensor能同时收到pico/device1/sensor和pico/device2/sensor的消息。上传数据的时候建议用设备类型/设备ID/数据类型这样的规范，方便后面接规则引擎做分流。

2.2 QoS 等级怎么选，retain 标志什么时候用

MQTT 定义了三个消息投递等级:

传感器节点我强烈建议 QoS 0 或者 QoS 1。QoS 0 性能最好，代价是网络抖动时数据会丢；QoS 1 会等 Broker 回 PUBACK，虽然可能重复但基本可靠。我这个项目用的是 QoS 1，因为发布频率不高（5 秒一条），开销可以忽略。

retain 标志也是个常用小开关。publish(topic, msg, retain=True)之后，Broker 会保存这条消息的"最新值"，任何新订阅者一上线就能立刻收到，而不是干等下一个发布周期。非常适合那种"设备状态"类数据，比如在线状态、当前温度；不适合普通日志流。

2.3 一条 PUBLISH 消息从板子到订阅端的七步

我自己在实际抓包中把这条链路拆成了七步，排查问题时按这个顺序走基本不会乱：

1. Pico 上的 MicroPython 代码调用client.publish()，json 字符串被编码成 bytes。
2. umqtt.simple 库按 MQTT 协议规定，给负载加上固定报头、Topic 字段、报文标识符，组成一个 PUBLISH 报文。
3. 报文通过 Wi-Fi 走 TCP 连接发送到 EMQX 的 1883 端口。
4. EMQX 先做协议解析，校验客户端是否已认证、是否有该 Topic 的发布权限。
5. Broker 根据 Topic 匹配订阅关系，找到所有匹配的订阅者。
6. 如果该主题开启了 retain，Broker 还会把消息存成最新值。
7. EMQX 把消息下发给每个订阅端，订阅端的 MQTT 客户端库收到后触发回调。

每一步都可能出问题：第 1 步代码写错、第 3 步网络不通、第 4 步账号不对、第 5 步 Topic 拼错，都会导致"看起来发了消息却收不到"。这也是为什么后文我会强调分段落验证，而不是一把梭。

3. EMQX 部署：三步搭好能收包的 Broker 并锁定访问权限

3.1 Docker 启动 EMQX：端口映射到底是哪几个

EMQX 的安装方式很多，官方提供 Docker 镜像、deb/rpm 包、还有二进制包。我图省事，直接用的 Docker。前提是你机器上已经有 Docker，没有的话照官方文档装一下也很简单。

我测试用的命令是：

docker run -d --name emqx \ -p 1883:1883 \ -p 8083:8083 \ -p 18083:18083 \ emqx/emqx:5.8.3

我要说清楚这三个端口各是干什么的，因为很多教程只扔命令不解释：

如果你后面要用 TLS 加密传输，还需要额外映射 8883（MQTT over TLS）。我跑通基础链路时先不开 TLS，毕竟 Pico 的 SSL 在 MicroPython 下挺吃内存，先把明文链路搞通再说安全。

启动之后浏览器访问http://localhost:18083，默认账号admin，密码public。

3.2 Dashboard 创建认证用户：从匿名访问到带锁 Broker

默认情况下 EMQX 允许匿名连接，也就是说只要网络能通，任何客户端都能发消息。这在纯局域网测试没问题，但一旦端口暴露到公网，情绪稳定的陌生人就能往你的 Topic 里灌垃圾。所以部署完第一件事就是加认证。

我自己的操作路径是：Dashboard 左侧菜单进入 Access Control -> Authentication，添加一个 Password-Based 的 Built-in Database 认证器，然后创建一组用户名和密码。我这里建的用户是pico，密码设成了随机生成的一串，而不是"123456"。

加了认证之后，匿名连接会被挡在门外。如果还想更严格一点，可以在 Authorization 里配置 ACL 规则，允许pico用户只对pico/#相关主题发布和订阅，并把默认未匹配策略从 allow 改成 deny。这样就算证书泄露，攻击者也只能碰这一个命名空间。

3.3 先用 MQTTX 手动验证 Broker 是否可用

在动 Pico 之前，强烈建议先在电脑上用 MQTTX 这个客户端工具把 Broker 验证一遍。MQTTX 是 EMQ 官方出的跨平台客户端，有图形界面，能极大降低调试心智负担。

我在 MQTTX 里新建一个连接，填入服务器地址192.168.1.100（EMQX 所在机器 IP）、端口1883，再填刚才创建的pico账号密码，点连接，看到 green light 就说明 Broker 本身没问题。然后订阅pico/sensor这个主题，在主界面手动发一条{"hello":"world"}测试消息，确认收得到。

这一步做的是"隔离变量"：先把服务端环境验证好，再上手嵌入式端。不然 Pico 连不上，你根本分不清是板子问题、网络问题还是 Broker 问题。

4. Pico 端准备：从裸板到带 Wi-Fi 的 MicroPython 环境

4.1 硬件清单与 DHT11 接线

我的硬件清单如下：

• 树莓派 Pico W 开发板一块
• DHT11 温湿度传感器一个
• 面包板、杜邦线若干
• 支持数据传输的 Micro-USB 线（有些线只能充电，无法通信，这个坑也常见）

DHT11 是一个很经典的数字温湿度传感器，虽然精度一般，但 MicroPython 内置dht模块，免去自己写时序协议的麻烦，非常适合做教学和原型验证。

接线很简单：

注意：DHT11 的数据引脚在长线传输时可能不稳定，建议杜邦线尽量短，或者干脆焊到排针上。我最初用一根 20cm 的杜邦线接 DHT11，偶尔会读不到数据，换成几厘米的短线后问题消失。

4.2 刷入带 Wi-Fi 的 MicroPython 固件

买回来的 Pico W 通常默认是空片或者出厂烧录的 MicroPython，但为了确保是支持 Wi-Fi 的版本，我还是重新刷了一遍固件。

流程是老三样：

1. 按住 Pico W 上的 BOOTSEL 按钮不放，同时用 USB 线把板子连到电脑。
2. 电脑上会弹出一个名为RPI-RP2的 U 盘。
3. 从 MicroPython 官网下载 RP2040 相关固件，这里一定要认准文件名里带w标识的版本，比如RPI_PICO_W。把它拖进 U 盘，板子会自动重启并运行新固件。

如果你给 Pico W 误刷了不带 W 的普通 Pico 固件，系统不会报错，但你在 REPL 里输入import network时，会发现 WLAN 相关功能缺失。这个坑我百思不得其解了很久，最后查固件版本才发现刷错了镜像。

我平时喜欢用 Thonny 这个 IDE 来连接板子写代码，它自带 MicroPython 的解释器交互界面，文件管理、代码上传、串口监视都齐全，对新手非常友好。

4.3 需要手动放进开发板的 umqtt.simple

新版本 MicroPython 固件本身不带 MQTT 客户端库，标准库里面只有网络相关的模块。所以还要把官方维护的umqtt.simple库塞进板子。

我推荐两种方式：

第一种，在 Thonny 的 Shell 里执行：

import mip mip.install("umqtt.simple")

第二种，去 micropython-lib 仓库找到umqtt.simple源码，下载simple.py，在 Thonny 的文件面板里把它存到 Pico 的/lib/umqtt/目录下，文件名必须是simple.py，这样代码里from umqtt.simple import MQTTClient才能正确导入。

我一开始不知道mip这回事，手动往板子里拖文件也成功了，后来发现mip一条命令就能搞定，省事很多。但是要注意：mip下载安装也需要网络，所以这个方法只有在板子已经连上 Wi-Fi 之后才能用。

5. 代码落地：Wi-Fi 重连、传感器采集、JSON 组装与 MQTT 发布

5.1 Wi-Fi 连接：超时与重连逻辑

MicroPython 连接 Wi-Fi 的代码本身不复杂，但直接按官方示例用 while 死等很容易挂死。我在实际测试中遇到过路由器重启、信号抖动导致连接失败的情况，所以给 Wi-Fi 连接加了超时控制。

def connect_wifi(): import network, time wlan = network.WLAN(network.STA_IF) wlan.active(True) if not wlan.isconnected(): print("connecting wifi...") wlan.connect(WIFI_SSID, WIFI_PASSWORD) for _ in range(30): if wlan.isconnected(): break time.sleep(0.5) if wlan.isconnected(): print("wifi ok:", wlan.ifconfig()) return True print("wifi status:", wlan.status()) return False

状态码wlan.status()在连接失败时能帮你定位问题，常见的有：WRONG_PASSWORD、NO_AP_FOUND、CONNECTION_FAIL，比看串口日志里的 raw 数字直观。我通常会在主循环里先判断 Wi-Fi 是否还连着，断了就重连，而不是直接去做 MQTT 操作。

5.2 时间戳的坑：NTP 校时不能省

一开始我发的 JSON 里带了"ts": time.time()这个字段，后来发现 Broker 收到的时间戳是 1970 年，或者跟真实时间差一大截。原因是 Pico 没有带电池的实时时钟芯片，每次上电后 RTC 时间默认从某个固定值开始计时。

解决办法是用 NTP 同步时间。MicroPython 提供了ntptime模块：

import ntptime, time ntptime.host = "ntp.aliyun.com" # 国内网络环境更快 ntptime.settime() print(time.localtime())

同步之后time.time()返回的就是 Unix 时间戳。这里有个小细节：ntptime.settime()设置的是 UTC 时间，如果你希望显示东八区本地时间，就自己加8 * 3600秒。

在内网完全隔离、访问不了外网 NTP 服务器的场景下，可以退而求其次，用自增序号或者开机相对秒数作为时间标识，至少能保证时序关系是对的。

5.3 JSON 组装与 MicroPython 的序列化差异

MicroPython 的json模块和 CPython 基本兼容，但有个差异值得注意：它没有完全暴露 CPython 的ensure_ascii参数。当你把中文字符串放进json.dumps()时，输出很可能变成了\u73af\u5883这样的转义序列。比如：

>>> import json >>> json.dumps({"location": "环境"}) '{"location": "\\u73af\\u5883"}'

这不是 bug，是一种保证跨平台兼容性的做法。接收端用标准json.loads()解析后会正确还原为中文，所以链路功能上没问题，只是你在看串口日志时不太直观。如果你确实想让下游看到原始 UTF-8 字符，就需要自己写一个小替换函数，或者换更轻量的第三方 JSON 库。我的建议是：真没必要，让数据接收方正常解析即可。

5.4 完整代码：发布循环 + 异常自愈

下面是我最终跑通的完整main.py，采集 DHT11 温湿度，组装成 JSON，发布到 EMQX：

import json import time import ntptime import network import dht from machine import Pin from umqtt.simple import MQTTClient # ========== 配置区，按你自己的环境改 ========== WIFI_SSID = "your_wifi_ssid" WIFI_PASSWORD = "your_wifi_password" MQTT_BROKER = "192.168.1.100" # EMQX 所在电脑的局域网 IP MQTT_PORT = 1883 MQTT_USER = "pico" MQTT_PASSWORD = "your_pico_password" MQTT_TOPIC = "pico/sensor" CLIENT_ID = "pico_w_01" DHT_PIN = 15 PUBLISH_INTERVAL = 5 # 发布间隔，单位秒 RETRY_INTERVAL = 5 # 连接失败后的重试间隔 # ========================================== sensor = dht.DHT11(Pin(DHT_PIN)) led = Pin("LED", Pin.OUT) def connect_wifi(): wlan = network.WLAN(network.STA_IF) wlan.active(True) if not wlan.isconnected(): print("connecting wifi...") wlan.connect(WIFI_SSID, WIFI_PASSWORD) for _ in range(30): if wlan.isconnected(): break time.sleep(0.5) if wlan.isconnected(): print("wifi ok:", wlan.ifconfig()) return True print("wifi failed, status:", wlan.status()) return False def sync_time(): try: ntptime.host = "ntp.aliyun.com" ntptime.settime() print("time synced:", time.localtime()) except Exception as e: print("ntp sync failed:", e) def read_sensor(): try: sensor.measure() return sensor.temperature(), sensor.humidity() except Exception as e: print("sensor read error:", e) return None, None def connect_mqtt(): client = MQTTClient( CLIENT_ID, MQTT_BROKER, port=MQTT_PORT, user=MQTT_USER, password=MQTT_PASSWORD, keepalive=60, ) client.connect() print("mqtt connected") return client def main(): if not connect_wifi(): print("exit: no wifi") return sync_time() client = None while True: if client is None: try: client = connect_mqtt() except Exception as e: print("mqtt connect error:", e) time.sleep(RETRY_INTERVAL) continue temp, hum = read_sensor() if temp is not None: payload = json.dumps({ "device_id": CLIENT_ID, "temperature": temp, "humidity": hum, "ts": time.time(), }) try: client.publish(MQTT_TOPIC, payload, qos=1) led.value(1) print("published:", payload) time.sleep(0.2) led.value(0) except Exception as e: print("publish error:", e) client = None continue time.sleep(PUBLISH_INTERVAL) main()

几个关键点说明一下：

• keepalive=60表示 60 秒内至少要有一条报文交互，否则 Broker 会认为客户端掉线并断开连接。我的发布间隔是 5 秒，小于 60，所以单纯发布就能保活。如果间隔大于 keepalive，就需要额外调用client.ping()。
• qos=1会让 Broker 返回确认包，如果网络抖动导致没收到确认，umqtt.simple 会抛异常，代码里捕获后置空 client，触发下一轮重连。
• 板载 LED 在每次成功发布后闪一下，用肉眼就能确认程序在正常工作，不用一直盯着串口。

6. 实测结果与高频坑位复盘：收不到包时的完整排查路径

6.1 用 MQTTX 订阅 topic 验证消息

代码上传到 Pico W 后，我第一次跑起来，MQTTX 里果然收到了消息，格式长这样：

{"device_id": "pico_w_01", "temperature": 26.0, "humidity": 62.0, "ts": 1723456789}

看到这条消息的时候，整个链路就算通了。如果你想在命令行验证，也可以用 mosquitto 客户端工具：

mosquitto_sub -h 192.168.1.100 -p 1883 -u pico -P your_password -t 'pico/sensor'

如果你更习惯用命令行，这种方式也很直接，而且方便嵌进脚本里做自动化断言。

6.2 连不上：换个顺序排查网络、端口、认证三个环节

收不到消息时，我的排查顺序基本固定：先看 Pico 串口日志里有没有wifi ok，再看有没有mqtt connected，最后看published。哪一步没出现，就往哪一步查。

• 第一步：网络。Pico 和 EMQX 机器必须在同一局域网，或者路由可达。用ping命令在电脑上确认 Broker IP 通不通；用 Pico 串口打印的wlan.ifconfig()确认板子拿到了合法 IP。
• 第二步：端口。在 EMQX 机器上执行netstat -tlnp | grep 1883确认端口在监听。如果 EMQX 跑在 Docker 里，不能只映射到127.0.0.1，要映射到0.0.0.0或局域网 IP，否则外部机器访问不到。
• 第三步：认证。确认 umqtt.simple 里的用户名密码和 Dashboard 里建的用户一致。EMQX 的认证失败日志在 Dashboard 的 Observation 页面里能直接看到，会明确告诉你client unauthorized之类的信息。

6.3 连接被断开：keepalive 与重连策略

我遇到过一种诡异情况：消息能正常发，但过几分钟 Pico 就断线，而且不会自动恢复。原因是当时的代码里没有重连机制，一旦网络波动导致底层 socket 断开，程序仍然以为自己还在线，后续 publish 直接抛异常。

umqtt.simple 的connect()只是在建立连接时认证一次，之后的连接维护需要你自己做。我的解决办法就在上面代码里：每次publish前把异常捕获住，只要抛异常，就把client置为None，主循环下一轮会重新connect_mqtt()。这样哪怕 Wi-Fi 断了又恢复、EMQX 重启了，程序都能自愈。

6.4 中文变 \uXXXX：是特性不是 bug

我在某个版本里把设备的物理位置写进了 JSON，比如"location": "客厅"，结果串口打印出来的是\u5ba2\u5385。一开始以为哪里搞坏了，后来确认这是 MicroPython json 模块的默认行为。

这条消息发到 EMQX，再用 MQTTX 订阅时，MQTTX 的 JSON 查看器显示的还是正常中文，因为标准解析器会还原转义字符。所以遇到这种情况不用慌，验证链路的最后一环即可。唯一的实际影响是：如果你在日志系统里直接搜"客厅"这个中文字符串，可能搜不到，要搜\u5ba2\u5385。这个细节在对接数据平台时要注意，但问题不大。

最后分享一个小习惯

项目跑通之后，我把整个链路拆成了三个独立验证层：第一层用 MQTTX 验证 Broker，第二层用 REPL 逐条命令验证 Wi-Fi 和 MQTT 库，第三层才写完整的 main.py。这样做最大的好处是：每层出问题时都能快速定位，不用在几十行代码里猜。以后你把这个组合换成 ESP32 或者其他支持 MicroPython 的板子时，这个分层验证的思路依然成立，代码只需要改网络配置和传感器引脚就行。如果你后续想让 Pico 不只是上报，还能接收下行指令去控制舵机，方式也一样——订阅一个控制 Topic，在回调里操作 GPIO，就能组成一个完整的闭环。