当前位置：首页 > news >正文

本地搭建MQTT服务器并进行设备控制（ESP32+MicroPython）

news 来源：原创 2025/4/21 12:12:13

本文用到的设备为esp32，固件是micropython。

1.mqtt服务端

我下的是开源版，下载 EMQX 开源版，免安装，以mac为例，下载解压后直接启动。

./emqx/bin/emqx start

若提示权限问题: sudo xattr -r -d com.apple.quarantine ./emqx

启动后，打开浏览器http://localhost:18083/ 进行登陆，用户名：admin, 密码：public

2.mqtt客户端（可选）

客户端用的是：MQTTX：全功能 MQTT 客户端工具

安装后打开，新建连接，配置如下：

host为你的本机ip地址，port固定1883，其他的随便填。

3.设备连接mqtt服务器

在设备上新建以下两个文件：

（1）umqttsimple.py

try:import usocket as socket
except:import socket
import ustruct as struct
from ubinascii import hexlifyclass MQTTException(Exception):passclass 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 1883self.client_id = client_idself.sock = Noneself.server = serverself.port = portself.ssl = sslself.ssl_params = ssl_paramsself.pid = 0self.cb = Noneself.user = userself.pswd = passwordself.keepalive = keepaliveself.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 = 0while 1:b = self.sock.read(1)[0]n |= (b & 0x7f) << shif not b & 0x80:return nsh += 7def set_callback(self, f):self.cb = fdef set_last_will(self, topic, msg, retain=False, qos=0):assert 0 <= qos <= 2assert topicself.lw_topic = topicself.lw_msg = msgself.lw_qos = qosself.lw_retain = retaindef 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 usslself.sock = ussl.wrap_socket(self.sock, **self.ssl_params)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 << 1if self.user is not None:sz += 2 + len(self.user) + 2 + len(self.pswd)msg[6] |= 0xC0if self.keepalive:assert self.keepalive < 65536msg[7] |= self.keepalive >> 8msg[8] |= self.keepalive & 0x00FFif 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) << 3msg[6] |= self.lw_retain << 5i = 1while sz > 0x7f:premsg[i] = (sz & 0x7f) | 0x80sz >>= 7i += 1premsg[i] = szself.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] == 0x02if resp[3] != 0:raise MQTTException(resp[3])return resp[2] & 1def disconnect(self):self.sock.write(b"\xe0\0")self.sock.close()def ping(self):self.sock.write(b"\xc0\0")def publish(self, topic, msg, retain=False, qos=0):pkt = bytearray(b"\x30\0\0\0")pkt[0] |= qos << 1 | retainsz = 2 + len(topic) + len(msg)if qos > 0:sz += 2assert sz < 2097152i = 1while sz > 0x7f:pkt[i] = (sz & 0x7f) | 0x80sz >>= 7i += 1pkt[i] = sz#print(hex(len(pkt)), hexlify(pkt, ":"))self.sock.write(pkt, i + 1)self._send_str(topic)if qos > 0:self.pid += 1pid = self.pidstruct.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:returnelif qos == 2:assert 0def subscribe(self, topic, qos=0):assert self.cb is not None, "Subscribe callback is not set"pkt = bytearray(b"\x82\0\0\0")self.pid += 1struct.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 Noneif res == b"":raise OSError(-1)if res == b"\xd0":  # PINGRESPsz = self.sock.read(1)[0]assert sz == 0return Noneop = res[0]if op & 0xf0 != 0x30:return opsz = 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 + 2if op & 6:pid = self.sock.read(2)pid = pid[0] << 8 | pid[1]sz -= 2msg = self.sock.read(sz)self.cb(topic, msg)if op & 6 == 2:pkt = bytearray(b"\x40\x02\0\0")struct.pack_into("!H", pkt, 2, pid)self.sock.write(pkt)elif op & 6 == 4:assert 0# 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()

（2）mqtt_client.py ，修改其中的wifi连接信息

import time
import network
from machine import Pin
from umqttsimple import MQTTClientWLAN_SSID = ''  # wifi名称
WLAN_PWD = ''    # wifi密码
MQTT_SERVER = '192.168.*.*'    # mqtt服务器ipdef do_connect():wlan = network.WLAN(network.STA_IF)wlan.active(True)if not wlan.isconnected():print('connecting to network...')wlan.connect(WLAN_SSID, WLAN_PWD)i=1while not wlan.isconnected():print("正在链接...{}".format(i))i+=1time.sleep(1)print('network config:', wlan.ifconfig())def sub_cb(topic, msg):# 回调函数，收到服务器消息后会调用这个函数print(topic, msg)if topic.decode("utf-8")=="ledctl" and msg.decode("utf-8")=="on":led_pin.value(1)if topic.decode("utf-8")=="ledctl" and msg.decode("utf-8")=="off":led_pin.value(0)#1.联网
do_connect()
# 2.创建mqtt
c=MQTTClient("esp32", MQTT_SERVER) #本设备作为一个MQTT客户端
c.set_callback(sub_cb) # 设置回调函数
c.connect() #建立连按
c.subscribe(b"ledctl")  #订阅ledct1这个主题，接收控制命令led_pin = Pin(2, Pin.OUT)while True:c.check_msg()time.sleep(1)

运行mqtt_client.py 连接mqtt服务器，监听消息。