手里这块树莓派 Pico 已经吃灰很久了最近把它翻出来做了个环境监测节点目标很简单用 MicroPython 采集温湿度通过 MQTT 协议把 JSON 消息发布到 EMQX Broker再在电脑上实时看到数据。整个过程走下来比我想象中要多绕好几个弯普通 Pico 根本没有 Wi-Fi、MicroPython 新版固件不自带 MQTT 库、EMQX 装好之后默认配置还有一堆安全细节要处理……这篇记录不是抄官方文档的流水账而是把我从翻出板子到最终稳定收到 JSON 数据全过程中踩过的坑、验证过的方案以及最终跑通的完整代码一起整理出来。适合手里有 Pico / Pico W、想做物联网数据上报、又不想一上来就啃 C 语言的读者。1. 选型逻辑为什么偏偏是 Pico MicroPython EMQX1.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 RAMMicroPython 解释器本身要占一部分跑大模型、高并发消息处理不现实。但作为传感器节点每 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最多一次消息可能丢失高频传感器数据丢一条无所谓1至少一次可能重复状态上报、控制指令2恰好一次性能开销最大计费、订单等必须精确的场景传感器节点我强烈建议 QoS 0 或者 QoS 1。QoS 0 性能最好代价是网络抖动时数据会丢QoS 1 会等 Broker 回 PUBACK虽然可能重复但基本可靠。我这个项目用的是 QoS 1因为发布频率不高5 秒一条开销可以忽略。retain 标志也是个常用小开关。publish(topic, msg, retainTrue)之后Broker 会保存这条消息的最新值任何新订阅者一上线就能立刻收到而不是干等下一个发布周期。非常适合那种设备状态类数据比如在线状态、当前温度不适合普通日志流。2.3 一条 PUBLISH 消息从板子到订阅端的七步我自己在实际抓包中把这条链路拆成了七步排查问题时按这个顺序走基本不会乱Pico 上的 MicroPython 代码调用client.publish()json 字符串被编码成 bytes。umqtt.simple 库按 MQTT 协议规定给负载加上固定报头、Topic 字段、报文标识符组成一个 PUBLISH 报文。报文通过 Wi-Fi 走 TCP 连接发送到 EMQX 的 1883 端口。EMQX 先做协议解析校验客户端是否已认证、是否有该 Topic 的发布权限。Broker 根据 Topic 匹配订阅关系找到所有匹配的订阅者。如果该主题开启了 retainBroker 还会把消息存成最新值。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我要说清楚这三个端口各是干什么的因为很多教程只扔命令不解释端口协议或用途1883MQTT 标准 TCP 接入端口Pico 走的就是这个8083MQTT over WebSocket给浏览器里的客户端用18083EMQX Dashboard 管理界面如果你后面要用 TLS 加密传输还需要额外映射 8883MQTT 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.100EMQX 所在机器 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 引脚连接目标VCCPico 的 3V3 引脚GNDPico 的 GNDDATAPico 的 GP15注意DHT11 的数据引脚在长线传输时可能不稳定建议杜邦线尽量短或者干脆焊到排针上。我最初用一根 20cm 的杜邦线接 DHT11偶尔会读不到数据换成几厘米的短线后问题消失。4.2 刷入带 Wi-Fi 的 MicroPython 固件买回来的 Pico W 通常默认是空片或者出厂烧录的 MicroPython但为了确保是支持 Wi-Fi 的版本我还是重新刷了一遍固件。流程是老三样按住 Pico W 上的 BOOTSEL 按钮不放同时用 USB 线把板子连到电脑。电脑上会弹出一个名为RPI-RP2的 U 盘。从 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发布到 EMQXimport 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, portMQTT_PORT, userMQTT_USER, passwordMQTT_PASSWORD, keepalive60, ) 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, qos1) 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()几个关键点说明一下keepalive60表示 60 秒内至少要有一条报文交互否则 Broker 会认为客户端掉线并断开连接。我的发布间隔是 5 秒小于 60所以单纯发布就能保活。如果间隔大于 keepalive就需要额外调用client.ping()。qos1会让 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就能组成一个完整的闭环。