大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



传统嵌入式MQTT开发的3大痛点，W55RP20一键解决：

1. 协议栈调试复杂：软件TCP/IP协议栈需手动移植、调试，占用MCU资源，易出现卡顿、内存溢出；W55RP20内置硬件TCP/IP协议栈，网络层处理完全由硬件完成，不占用CPU资源，无需调试协议栈。

2. 硬件接线繁琐：外接网络模块需手动焊接、调试接口时序，新手易出错；W55RP20板载以太网接口，无需额外外接模块，即插即用，大幅简化接线操作。

3. 入门门槛高：MQTT客户端连接、心跳保活等细节需手动编写，调试周期长；本文提供可直接运行的完整代码，注释清晰，配合10分钟分步操作，新手也能快速出效果，依托MicroPython免编译特性，代码修改即时生效、调试实时反馈，快速获得成就感。

本节课核心目：

· 完成W55RP20与电脑的硬件连接、开发环境配置

· 理解MQTT协议核心原理与极简工作流程

· 烧录MQTT客户端代码，实现与公共MQTT服务器的连接

· 完成主题订阅、消息发布与回环通信测试

· 掌握常见故障的快速排查方法

本实战系列共16篇，循序渐进覆盖W55RP20的基础联网、MQTT通信、工业协议对接等全场景，建议收藏本系列，跟随教程逐步学习，所有代码均会同步更新至官方Gitee仓库。

1. 准备工作

1.1 软件准备

1.2 硬件准备

2. 快速确认固件与开发环境

3. 硬件连接（1分钟）

4. MQTT协议极简解析

5. 核心代码复制与烧录

5.1 依赖库说明

5.2 完整代码（含详细中文注释）

6. 运行结果与MQTT通信验证

6.1 串口输出结果

6.2 MQTT回环通信验证

7. 常见问题一站式排查

8. W55RP20 MQTT开发核心优势

9. 系列预告与资源获取

9.1 系列预告

9.2 资源获取

W55RP20 开发板 × 1

Micro USB 数据线（必须支持数据传输，不能使用纯充电线） × 1

标准网线 × 1

说明：W55RP20开发板已板载以太网接口和WIZnet硬件TCP/IP协议栈，无需额外外接网络模块、无需焊接飞线，即插即用，大幅降低接线错误和硬件故障概率，完美适配10分钟快速实战需求。

1. 确认已烧录 W55RP20 专属 MicroPython 固件（若未烧录，参考系列第1篇教程，30秒即可完成烧录）；

2. 打开Thonny软件，点击顶部菜单栏「运行」→「配置解释器」，确认解释器选择「MicroPython (通用)」，端口选择W55RP20对应的串口（通常显示为Board CDC @ COMx）；

3. 确认umqttsimple.py库已保存至W55RP20开发板（将库文件拖拽至Thonny左侧的开发板文件列表即可）。

W55RP20连接极其简单，仅需2步，无需复杂接线：

1. 使用 Micro USB 数据线连接W55RP20开发板与电脑（用于供电、代码烧录和串口调试）；

2. 使用网线连接W55RP20的以太网接口与路由器的LAN口（确保路由器已开启DHCP功能，无需手动配置IP）。

【硬件预留】此处插入W55RP20硬件连接示意图（电脑+开发板+路由器连接）

MQTT是一种轻量级物联网消息传输协议，基于发布/订阅模式，适用于硬件资源有限的嵌入式设备和带宽有限的网络环境，广泛应用于物联网、智能硬件、工业控制等场景，核心优势是轻量、高效、可靠、易实现，最小控制消息仅需两个数据字节，消息头精简，可优化网络带宽。

极简工作流程（4步）：

1. 客户端（W55RP20）连接到MQTT服务器（本文使用免费公共服务器，无需注册）；

2. 客户端订阅指定主题（如/W55RP20/sub），用于接收消息；

3. 客户端（或其他设备）向指定主题发布消息；

4. 所有订阅该主题的客户端，均可接收并处理消息。

本文实战使用免费公共MQTT服务器（broker-cn.emqx.io，端口1883），无需注册，直接连接即可使用，进一步节省时间，助力10分钟完成实战。

核心依赖 umqttsimple.py 库，用于实现MQTT客户端的连接、订阅、发布等基础功能，该库是MicroPython生态中常用的轻量级MQTT库，体积小、适配性强，无需额外修改，直接保存至W55RP20即可使用，可通过MicroPython官方仓库下载或直接复制本文提供的完整代码。

第二步：将umqttsimple.py库文件保存到开发板 中。

第三步：运行程序，并打开MQTTX ，连接相同的服务器，然后订阅主题为开发板的发布主题，发布主题为开发板的订阅主题。下面有图

第四步：在MQTTX发送消息观察回环测试效果。

注意：因为MicroPython的print函数是启用了stdout缓冲的，所以有时候并不会第一时间打印出内容。

mqttsimple.py代码：

#mqtt.py file import network import time try: from machine import Pin, WIZNET_PIO_SPI except ImportError: WIZNET_PIO_SPI = None Pin = None _DEFAULTS = { "w5100s-evb-pico": {}, "w5500-evb-pico": {}, "w6100-evb-pico": {}, "w5100s-evb-pico2": {}, "w5500-evb-pico2": {}, "w6100-evb-pico2": {}, "w55rp20-evb-pico": , "w6300-evb-pico": , "w6300-evb-pico2": , } _AUTO = { "w5100s-evb-pico", "w5500-evb-pico", "w6100-evb-pico", "w5100s-evb-pico2","w5500-evb-pico2","w6100-evb-pico2", } _SINGLE = {"w55rp20-evb-pico"} _QSPI = {"w6300-evb-pico", "w6300-evb-pico2"} def _pin(x): return x if isinstance(x, Pin) else Pin(x) def wiznet(board, *, dhcp=True, spi=None, cs=None, reset=None, kw): board = board.strip().lower() if board not in _DEFAULTS: raise ValueError("Unsupported board: {}".format(board)) cfg = _DEFAULTS[board].copy() cfg.update(kw) if spi is not None: if cs is None or reset is None: raise ValueError("When passing custom spi, also pass cs and reset") nic = network.WIZNET6K(spi, cs, reset) else: if board in _AUTO: nic = network.WIZNET6K() elif board in _SINGLE: if WIZNET_PIO_SPI is None or Pin is None: raise RuntimeError("WIZNET_PIO_SPI/Pin not available on this port") required = ["sck", "cs", "mosi", "miso", "reset"] missing = [k for k in required if k not in cfg] if missing: raise ValueError("Missing pins for W55RP20 single-SPI: " + ", ".join(missing)) spi = WIZNET_PIO_SPI( baudrate=cfg.get("baudrate", ), sck=_pin(cfg["sck"]), cs=_pin(cfg["cs"]), mosi=_pin(cfg["mosi"]), miso=_pin(cfg["miso"]), ) nic = network.WIZNET6K(spi, _pin(cfg["cs"]), _pin(cfg["reset"])) elif board in _QSPI: if WIZNET_PIO_SPI is None or Pin is None: raise RuntimeError("WIZNET_PIO_SPI/Pin not available on this port") for k in ["sck","cs","io0","io1","io2","io3"]: if k not in cfg: raise ValueError("Missing pin '{}' for W6300 QSPI".format(k)) spi = WIZNET_PIO_SPI( baudrate=cfg.get("baudrate", ), sck=_pin(cfg["sck"]), cs=_pin(cfg["cs"]), io0=_pin(cfg["io0"]), io1=_pin(cfg["io1"]), io2=_pin(cfg["io2"]), io3=_pin(cfg["io3"]), ) nic = network.WIZNET6K(spi, _pin(cfg["cs"]), _pin(cfg.get("reset", cfg["cs"]))) else: raise ValueError("Unexpected board mapping") try: nic.active(True) except AttributeError: pass if dhcp: try: nic.ifconfig("dhcp") except Exception: pass else: ip = cfg.get("ip"); sn = cfg.get("sn"); gw = cfg.get("gw"); dns = cfg.get("dns", gw or "8.8.8.8") if not (ip and sn and gw): raise ValueError("Static mode requires ip/sn/gw") nic.ifconfig((ip, sn, gw, dns)) while not nic.isconnected(): print("Waiting for the network to connect...") time.sleep(1) print("MAC Address:", ":".join("%02x" % b for b in nic.config("mac"))) print("IP Address:", nic.ifconfig()[0]) return nic from umqttsimple import MQTTClient from machine import Timer import machine #mqtt config mqtt_params = {} mqtt_params['url'] = 'broker.emqx.io' mqtt_params['port'] = 1883 mqtt_params['clientid'] = 'W55RP20' mqtt_params['pubtopic'] = '/W55RP20/pub' mqtt_params['subtopic'] = '/W55RP20/sub' mqtt_params['pubqos'] = 0 mqtt_params['subqos'] = 0 timer_1s_count = 0 tim = Timer() client = None def w5x00_init(): nic = wiznet("w55rp20-evb-pico", dhcp=True) print("IP :",nic.ifconfig()[0]) print("Subnet Mask:",nic.ifconfig()[1]) print("Gateway :",nic.ifconfig()[2]) print("DNS :",nic.ifconfig()[3]," ") return nic def sub_cb(topic, msg): topic = topic.decode('utf-8') msg = msg.decode('utf-8') if topic == mqtt_params['subtopic']: global client print(" topic:",topic," recv:", msg) client.publish(mqtt_params['pubtopic'],msg,qos=mqtt_params['pubqos']) print(' topic:',mqtt_params['pubtopic'],' send:',msg) def mqtt_connect(): global client # 每次重连换不同ID，防止服务器拒绝 from time import time new_id = "W55RP20_" + str(time()) client = MQTTClient(new_id, mqtt_params['url'], mqtt_params['port'], keepalive=60) client.connect() print('Connected to %s MQTT Broker'%(mqtt_params['url'])) return client def reconnect(): print('MQTT 连接断开，正在重新连接...') time.sleep(3) def tick(timer): global timer_1s_count global client timer_1s_count += 1 if timer_1s_count >= 30: timer_1s_count = 0 try: client.ping() except: pass def subscribe(client): client.set_callback(sub_cb) client.subscribe(mqtt_params['subtopic'],mqtt_params['subqos']) print('subscribed to %s'%mqtt_params['subtopic']) def main(): global client print("W55RP20 chip MQTT example") # 网络只初始化一次 w5x00_init() # ========== 核心：无限循环 + 出错自动重连 ========== while True: try: client = mqtt_connect() subscribe(client) tim.init(freq=1, callback=tick) # 正常运行 while True: client.wait_msg() except Exception as e: print("检测到错误，即将重连：", e) try: client.disconnect() except: pass time.sleep(2) if __name__ == "__main__": main() 

补充umqttsimple.py库代码（复制保存至W55RP20开发板，与mqtt.py同级）：

#umqttsimple.py file import usocket as socket import ustruct as struct from ubinascii import hexlify class MQTTException(Exception): pass class MQTTClient: def __init__( self, client_id, server, port=0, user=None, password=None, keepalive=0, ssl=False, ssl_params={}, ): if port == 0: port = 8883 if ssl else 1883 self.client_id = client_id self.sock = None self.server = server self.port = port self.ssl = ssl self.ssl_params = ssl_params self.pid = 0 self.cb = None self.user = user self.pswd = password self.keepalive = keepalive self.lw_topic = None self.lw_msg = None self.lw_qos = 0 self.lw_retain = False def _send_str(self, s): self.sock.write(struct.pack("!H", len(s))) self.sock.write(s) def _recv_len(self): n = 0 sh = 0 while 1: b = self.sock.read(1)[0] n |= (b & 0x7F) << sh if not b & 0x80: return n sh += 7 def set_callback(self, f): self.cb = f def set_last_will(self, topic, msg, retain=False, qos=0): assert 0 <= qos <= 2 assert topic self.lw_topic = topic self.lw_msg = msg self.lw_qos = qos self.lw_retain = retain def connect(self, clean_session=True): self.sock = socket.socket() addr = socket.getaddrinfo(self.server, self.port)[0][-1] self.sock.connect(addr) if self.ssl: import ussl self.sock = ussl.wrap_socket(self.sock, self.ssl_params) premsg = bytearray(b"x") msg = bytearray(b"x04MQTTx04x0200") sz = 10 + 2 + len(self.client_id) msg[6] = clean_session << 1 if self.user is not None: sz += 2 + len(self.user) + 2 + len(self.pswd) msg[6] |= 0xC0 if self.keepalive: assert self.keepalive < 65536 msg[7] |= self.keepalive >> 8 msg[8] |= self.keepalive & 0x00FF if self.lw_topic: sz += 2 + len(self.lw_topic) + 2 + len(self.lw_msg) msg[6] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3 msg[6] |= self.lw_retain << 5 i = 1 while sz > 0x7F: premsg[i] = (sz & 0x7F) | 0x80 sz >>= 7 i += 1 premsg[i] = sz self.sock.write(premsg, i + 2) self.sock.write(msg) # print(hex(len(msg)), hexlify(msg, ":")) self._send_str(self.client_id) if self.lw_topic: self._send_str(self.lw_topic) self._send_str(self.lw_msg) if self.user is not None: self._send_str(self.user) self._send_str(self.pswd) resp = self.sock.read(4) assert resp[0] == 0x20 and resp[1] == 0x02 if resp[3] != 0: raise MQTTException(resp[3]) return resp[2] & 1 def disconnect(self): self.sock.write(b"xe00") self.sock.close() def ping(self): self.sock.write(b"xc00") def publish(self, topic, msg, retain=False, qos=0): pkt = bytearray(b"x30000") pkt[0] |= qos << 1 | retain sz = 2 + len(topic) + len(msg) if qos > 0: sz += 2 assert sz <  i = 1 while sz > 0x7F: pkt[i] = (sz & 0x7F) | 0x80 sz >>= 7 i += 1 pkt[i] = sz # print(hex(len(pkt)), hexlify(pkt, ":")) self.sock.write(pkt, i + 1) self._send_str(topic) if qos > 0: self.pid += 1 pid = self.pid struct.pack_into("!H", pkt, 0, pid) self.sock.write(pkt, 2) self.sock.write(msg) if qos == 1: while 1: op = self.wait_msg() if op == 0x40: sz = self.sock.read(1) assert sz == b"x02" rcv_pid = self.sock.read(2) rcv_pid = rcv_pid[0] << 8 | rcv_pid[1] if pid == rcv_pid: return elif qos == 2: assert 0 def subscribe(self, topic, qos=0): assert self.cb is not None, "Subscribe callback is not set" pkt = bytearray(b"x82000") self.pid += 1 struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid) # print(hex(len(pkt)), hexlify(pkt, ":")) self.sock.write(pkt) self._send_str(topic) self.sock.write(qos.to_bytes(1, "little")) while 1: op = self.wait_msg() if op == 0x90: resp = self.sock.read(4) # print(resp) assert resp[1] == pkt[2] and resp[2] == pkt[3] if resp[3] == 0x80: raise MQTTException(resp[3]) return # Wait for a single incoming MQTT message and process it. # Subscribed messages are delivered to a callback previously # set by .set_callback() method. Other (internal) MQTT # messages processed internally. def wait_msg(self): res = self.sock.read(1) # self.sock.setblocking(True) if res is None: return None if res == b"": raise OSError(-1) if res == b"xd0": # PINGRESP sz = self.sock.read(1)[0] assert sz == 0 return None op = res[0] if op & 0xF0 != 0x30: return op sz = self._recv_len() topic_len = self.sock.read(2) topic_len = (topic_len[0] << 8) | topic_len[1] topic = self.sock.read(topic_len) sz -= topic_len + 2 if op & 6: pid = self.sock.read(2) pid = pid[0] << 8 | pid[1] sz -= 2 msg = self.sock.read(sz) self.cb(topic, msg) if op & 6 == 2: pkt = bytearray(b"x40x0200") struct.pack_into("!H", pkt, 2, pid) self.sock.write(pkt) elif op & 6 == 4: assert 0 return op # Checks whether a pending message from server is available. # If not, returns immediately with None. Otherwise, does # the same processing as wait_msg. def check_msg(self): # self.sock.setblocking(False) return self.wait_msg() 

补充wiznet_init.py库代码（复制保存至W55RP20开发板，与mqtt.py同级）：

import network import time try: from machine import Pin, WIZNET_PIO_SPI except ImportError: WIZNET_PIO_SPI = None Pin = None _DEFAULTS = { # Auto-construct boards (no explicit PIO SPI) "w5100s-evb-pico": {}, "w5500-evb-pico": {}, "w6100-evb-pico": {}, "w5100s-evb-pico2": {}, "w5500-evb-pico2": {}, "w6100-evb-pico2": {}, # W55RP20 — single SPI (PIO SPI) "w55rp20-evb-pico": , # W6300 — QSPI QUAD(io0..io3) "w6300-evb-pico": , "w6300-evb-pico2": , } _AUTO = { "w5100s-evb-pico", "w5500-evb-pico", "w6100-evb-pico", "w5100s-evb-pico2","w5500-evb-pico2","w6100-evb-pico2", } _SINGLE = {"w55rp20-evb-pico"} # PIO single-SPI _QSPI = {"w6300-evb-pico", "w6300-evb-pico2"} def _pin(x): return x if isinstance(x, Pin) else Pin(x) def wiznet(board, *, dhcp=True, spi=None, cs=None, reset=None, kw): board = board.strip().lower() if board not in _DEFAULTS: raise ValueError("Unsupported board: {}".format(board)) cfg = _DEFAULTS[board].copy() cfg.update(kw) # Manual override path: if spi is provided, use it directly if spi is not None: if cs is None or reset is None: raise ValueError("When passing custom spi, also pass cs and reset") nic = network.WIZNET6K(spi, cs, reset) else: if board in _AUTO: nic = network.WIZNET6K() elif board in _SINGLE: if WIZNET_PIO_SPI is None or Pin is None: raise RuntimeError("WIZNET_PIO_SPI/Pin not available on this port") required = ["sck", "cs", "mosi", "miso", "reset"] missing = [k for k in required if k not in cfg] if missing: raise ValueError("Missing pins for W55RP20 single-SPI: " + ", ".join(missing)) spi = WIZNET_PIO_SPI( baudrate=cfg.get("baudrate", ), sck=_pin(cfg["sck"]), cs=_pin(cfg["cs"]), mosi=_pin(cfg["mosi"]), miso=_pin(cfg["miso"]), ) nic = network.WIZNET6K(spi, _pin(cfg["cs"]), _pin(cfg["reset"])) elif board in _QSPI: if WIZNET_PIO_SPI is None or Pin is None: raise RuntimeError("WIZNET_PIO_SPI/Pin not available on this port") for k in ["sck","cs","io0","io1","io2","io3"]: if k not in cfg: raise ValueError("Missing pin '{}' for W6300 QSPI".format(k)) spi = WIZNET_PIO_SPI( baudrate=cfg.get("baudrate", ), sck=_pin(cfg["sck"]), cs=_pin(cfg["cs"]), io0=_pin(cfg["io0"]), io1=_pin(cfg["io1"]), io2=_pin(cfg["io2"]), io3=_pin(cfg["io3"]), ) nic = network.WIZNET6K(spi, _pin(cfg["cs"]), _pin(cfg.get("reset", cfg["cs"]))) else: raise ValueError("Unexpected board mapping") # Bring up (if supported) try: nic.active(True) except AttributeError: pass if dhcp: try: nic.ifconfig("dhcp") except Exception: pass else: ip = cfg.get("ip"); sn = cfg.get("sn"); gw = cfg.get("gw"); dns = cfg.get("dns", gw or "8.8.8.8") if not (ip and sn and gw): raise ValueError("Static mode requires ip/sn/gw") nic.ifconfig((ip, sn, gw, dns)) while not nic.isconnected(): print("Waiting for the network to connect...") time.sleep(1) print("MAC Address:", ":".join("%02x" % b for b in nic.config("mac"))) print("IP Address:", nic.ifconfig()) return nic 

在Thonny中点击运行按钮（或按F5键），Shell窗口会输出以下内容，说明网络初始化和MQTT连接成功：

说明：若打印“Waiting for the network to connect.....”，属于正常现象，等待1-3秒即可获取IP；若一直无法获取IP，参考“常见问题排查”部分。

 MPY: soft reboot W55RP20 chip MQTT example Waiting for the network to connect... Waiting for the network to connect... Waiting for the network to connect... Waiting for the network to connect... MAC Address: 02:90:86:88:4d:56 IP Address: 192.168.1.118 IP : 192.168.1.118 Subnet Mask: 255.255.255.0 Gateway : 192.168.1.1 DNS : 202.96.134.33 Connected to broker.emqx.io MQTT Broker subscribed to /W55RP20/sub

使用工具测试，步骤如下（不会的照着下面的图来）：

mqttx网址：MQTTX：全功能 MQTT 客户端工具

1. 打开MQTTX，点击“+”号，创建新连接：

- 名称：自定义（如W55RP20测试）

- 服务器：broker-cn.emqx.io

- 端口：1883

- 其他参数默认，点击“连接”；

2. 订阅W55RP20的发布主题：/W55RP20/pub；可以自己定义

3. 向W55RP20的订阅主题：/W55RP20/sub 发送消息（如“W55RP20 MQTT测试”）；可以自己定义

4. 观察两个地方的反馈：

- Thonny Shell窗口：打印收到的消息，以及“消息回环成功”提示；

- MQTTX工具：收到W55RP20回环发布的消息（与发送的消息一致）。

至此，W55RP20的MQTT通信实战完成

为了让你更直观地了解 W55RP20 的价值，我们对比了目前主流的三种嵌入式以太网方案：

W55RP20 芯片集成以太网功能，结合其工业级稳定性，非常适合以下应用场景：

下一篇教程我们将讲解 W55RP20 MicroPython开发下的 MQTT+阿里云实现上传数据，带你了解连接建立、心跳包维护、异常重连等关键机制，掌握嵌入式以太网通信的基础能力。