使用 Docker 搭建 EMQX，并通过 MicroPython 实现 MQTT 控制 ESP32 LED

使用 Docker 搭建 EMQX，并通过 MicroPython 实现 MQTT 控制 ESP32 LED

在物联网项目中，MQTT是最常用的消息通信协议之一。本文将通过一个完整示例，演示如何：

1. 使用Docker 快速部署 EMQX
2. 解决端口冲突问题
3. 使用MicroPython作为 MQTT 客户端
4. 通过 MQTT 消息远程控制 ESP32 LED

一、使用 Docker 快速启动 EMQX

1. 基础启动命令

最基础的 EMQX 容器启动方式如下：

dockerrun-itd\--nameemqx\-p1883:1883\-p8083:8083\-p8084:8084\-p8883:8883\-p18083:18083\emqx/emqx

容器启动成功后会返回类似 ID：

185326ea778710a12c77ad3b3e24800ce7591c30f4feb0196742bfd70f655d14

这样：

• 宿主机：1884
• 容器内部：1883（MQTT 默认端口）

二、访问 EMQX 控制台

浏览器访问（HTTPS）：

https://服务器IP:18083

默认账号密码：

用户名：admin 密码：public

控制台界面如下：

三、MicroPython MQTT 客户端实现

ESP32 使用MicroPython，MQTT 客户端代码基于simple.py实现。

1. MQTT 客户端库（simple.py）

文件名：simple.py

importusocketassocketimportustructasstructfromubinasciiimporthexlifyclassMQTTException(Exception):passclassMQTTClient:def__init__(self,client_id,server,port=0,user=None,password=None,keepalive=0,ssl=None,):ifport==0:port=8883ifsslelse1883self.client_id=client_id self.sock=Noneself.server=server self.port=port self.ssl=ssl self.pid=0self.cb=Noneself.user=user self.pswd=password self.keepalive=keepalive self.lw_topic=Noneself.lw_msg=Noneself.lw_qos=0self.lw_retain=Falsedef_send_str(self,s):self.sock.write(struct.pack("!H",len(s)))self.sock.write(s)def_recv_len(self):n=0sh=0while1:b=self.sock.read(1)[0]n|=(b&0x7F)<<shifnotb&0x80:returnn sh+=7defset_callback(self,f):self.cb=fdefset_last_will(self,topic,msg,retain=False,qos=0):self.lw_topic=topic self.lw_msg=msg self.lw_qos=qos self.lw_retain=retaindefconnect(self,clean_session=True):self.sock=socket.socket()addr=socket.getaddrinfo(self.server,self.port)[0][-1]self.sock.connect(addr)premsg=bytearray(b"\x10\0\0\0\0\0")msg=bytearray(b"\x04MQTT\x04\x02\0\0")sz=10+2+len(self.client_id)msg[6]=clean_session<<1ifself.user:sz+=2+len(self.user)+2+len(self.pswd)msg[6]|=0xC0i=1whilesz>0x7F:premsg[i]=(sz&0x7F)|0x80sz>>=7i+=1premsg[i]=sz self.sock.write(premsg,i+2)self.sock.write(msg)self._send_str(self.client_id)ifself.user:self._send_str(self.user)self._send_str(self.pswd)resp=self.sock.read(4)ifresp[3]!=0:raiseMQTTException(resp[3])defsubscribe(self,topic,qos=0):self.cbisnotNonepkt=bytearray(b"\x82\0\0\0")self.pid+=1struct.pack_into("!BH",pkt,1,2+2+len(topic)+1,self.pid)self.sock.write(pkt)self._send_str(topic)self.sock.write(qos.to_bytes(1,"little"))defcheck_msg(self):self.sock.setblocking(False)try:res=self.sock.read(1)except:returnifres==b"\x30":self.sock.read(1)tlen=self.sock.read(2)tlen=(tlen[0]<<8)|tlen[1]topic=self.sock.read(tlen)msg=self.sock.read()self.cb(topic,msg)

四、ESP32 主程序（控制 LED）

文件名：main.py

fromsimpleimportMQTTClientfrommachineimportPinimportnetwork led=Pin(2,Pin.OUT)SSID="wifi名称"PASSWORD="wifi密码"wlan=network.WLAN(network.STA_IF)wlan.active(True)ifnotwlan.isconnected():wlan.connect(SSID,PASSWORD)whilenotwlan.isconnected():passprint("Network config:",wlan.ifconfig())SERVER="192.168.31.219"PORT=1884CLIENT_ID="my_micropython_client"defsub(topic,msg):print(topic,msg)iftopic==b"led002":ifmsg==b"on":led.value(0)elifmsg==b"off":led.value(1)client=MQTTClient(CLIENT_ID,SERVER,PORT,"test","123456")client.set_callback(sub)client.connect()client.subscribe(b"led002")whileTrue:client.check_msg()

五、测试效果

通过 EMQX 控制台或其他 MQTT 客户端：

• Topic：led002

• Payload：

  • on→ LED 亮
  • off→ LED 灭

设备成功订阅并响应：