物联网开发笔记（55）- 使用Micropython开发ESP32开发板之MQTT(消息队列遥测传输)实战

一、目的

        这一节我们学习如何使用我们的ESP32开发板来学习MQTT。

二、环境

        ESP32客户端 + MQTT服务器 + Thonny IDE + 电脑 + 手机

三、MQTT服务器的安装

        MQTT具体是啥？大家自行百度，此处不再赘述。我找到一个介绍的网站，大家也可以去这里看一下：

1-1 MQTT是什么 – 太极创客

这里我和大家介绍下安装方法，首先下载（此处使用的版本是V5.0.11）：

下载 EMQX

下载完成后，解压，然后打开CMD命令行，输入如下命令"emqx start"，就可以启动：

 具体的使用方法，可参考官方的使用指南：

快速开始 | EMQX 5.0 文档

我们接着往下讲，上面资料你后面再看。打开浏览器，输入http://localhost:18083/

输入默认用户名admin和密码public，登录进去：

 登录进去会强制让你更改密码，你设置好新密码后，就可以操作了。

四、ESP32客户端程序

        我们在电脑上安装了MQTT的服务器，下面我们在ESP32开发板上建立客户端。大概步骤如下：

1. esp32链接上wifi（注意与当前电脑链接到同一个局域网下，即同一个wifi路由器）
2. esp32上需要下载一个mqtt的库
3. 我们自己编写一个.py文件调用上面第2步的mqtt库，从而使用mqtt实现客户端的功能

        第2步骤的mqtt库文件，名字叫做umqttsimple.py，内容如下，不要做任何修改复制到esp32保存即可。

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"\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 << 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"\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 | retain
        sz = 2 + len(topic) + len(msg)
        if qos > 0:
            sz += 2
        assert sz < 2097152
        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"\x82\0\0\0")
        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"\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()

第1和第3步要编写的自己的代码，文件名字MQTT_Test.py，内容如下

import time
import network
from umqttsimple import MQTTClient

# ESP32连接无线网
def do_connect():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print('connecting to network...')
        wlan.connect('TP-LINK_33E4', '2019AP3029')  # WIFI名字和密码
        i = 1
        while not wlan.isconnected():
            print("正在链接中...{}".format(i))
            i += 1
            time.sleep(1)
    print('network config:', wlan.ifconfig())


def sub_cb(topic, msg): # 回调函数，收到服务器消息后会调用这个函数
    print(topic, msg)


# 1. 联网
do_connect()
# 2. 创建mqtt
c = MQTTClient("umqtt_client", "192.168.0.102")  # 建立一个MQTT客户端，这个IP是电脑的IP地址
c.set_callback(sub_cb)  # 设置回调函数
c.connect()  # 建立连接
c.subscribe(b"ledctl")  # 监控ledctl这个通道，接收控制命令

while True:
    c.check_msg()
    time.sleep(1)

运行程序：

这里我们完成了ESP32充当MQTT客户端，自己的电脑充当MQTT服务器。

五、电脑上安装MQTT客户端

        MQTT是C/S架构模式，我们已经在电脑上安装了MQTT服务器，在ESP32编写了MQTT客户端程序。下面，我们将要在另外一台设备（电脑）上也运行MQTT客户端，最终实现让这个新的客户端发送控制命令给ESP32，最终实现控制ESP32上的LED的亮灭。

 我们需要在另外一台电脑上安装MQTT客户端，下载链接如下：

MQTT X：跨平台 MQTT 5.0 桌面客户端工具

 然后按照默认安装即可。

我这里为了方便，把客户端和服务端直接安装在同一台电脑上哈。。。

安装完成后，我们打开MQTT客户端：

 然后我们连接MQTT服务器。注意：

首先要确保MQTT服务器是运行的，如果电脑重启，那么上次运行的MQTT服务器就不会自动运行，需要用上面相同的方式将MQTT服务器运行起来，所以先用浏览器方式http://localhost:18083/看看能否打开，如果打开就说明MQTT服务器运行正常，否则需要手动开启MQTT服务器。

我们选择 New Connection：

，填写如下信息，注意这个IP地址是MQTT服务器的地址：

然后会提示连接成功：

 然后我们在MQTT服务端，可以看到连接成功：

 我们这里只运行了一个客户端，所以只有一个，ESP32客户端没有运行。

六、ESP32上运行新的客户端程序

将前面的ESP32 MQTT客户端的代码，加上控制LED的代码：

import time
from machine import Pin  
import network
from umqttsimple import MQTTClient

# 创建WIFI连接函数
def do_connect():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print('connecting to network...')
        wlan.connect('名字', '密码')  # WIFI名字和密码
        i = 1
        while not wlan.isconnected():
            print("正在链接中...{}".format(i))
            i += 1
            time.sleep(1)
    print('network config:', wlan.ifconfig())


def sub_cb(topic, msg): # 回调函数，收到服务器消息后会调用这个函数
    print(topic, msg)
    if topic.decode("utf-8") == "BlueLED" and msg.decode("utf-8") == "on":
        led_pin.value(1)
    elif topic.decode("utf-8") == "BlueLED" and msg.decode("utf-8") == "off":
        led_pin.value(0)

# 1. 联网
do_connect()

# 2. 创建mqt
c = MQTTClient("umqtt_client", "192.168.0.102")  # 建立一个MQTT客户端。IP为MQTT服务器的IP
c.set_callback(sub_cb)  # 设置回调函数
c.connect()  # 建立连接
c.subscribe(b"BlueLED")  # 监控ledctl这个通道，接收控制命令

# 3. 创建LED对应Pin对象
led_pin = Pin(2, Pin.OUT)  # 通过GPIO口2控制ESP32板载的蓝色LED灯

while True:
    c.check_msg()
    time.sleep(1)

我们运行上面代码，在MQTT服务端，会看到连接成功：

 此时ESP32 板载蓝色LED灯还是不亮的哦~

七、在电脑的MQTT客户端上发送消息

当前发送的主题是BlueLED（与上述代码中的ESP32订阅的消息一样），发送的内容是on。

 然后我们就会看到ESP32板载LED灯亮了。

然后我们再发送off，会看在灯灭了

 

 七、在手机上的MQTT客户端上发送消息

         以上我们实现了再电脑的MQTT客户端发送消息，控制ESP32客户端的蓝色LED灯。那么下面我们讲解怎么在手机上运行MQTT服务端来控制ESP32客户端的蓝色LED灯。

        此处我使用安卓系统的小米10手机为大家演示，大家现在手机自带的应用商店中搜索"MQTT调试器"并安装，安装成功后打开。我安装时版本是1.0.8。

        在开始之前，我们检查下我们的MQTT服务器应该开启，没开启的请查看本文上方的章节。

我们MQTT服务器会成功连接3个客户端，一个是电脑客户端，一个是手机客户端，一个是ESP32客户端。如下图MQTT服务器显示：

 好啦，我们在手机上打开MQTT调试器，按照下面步骤操作：

 

 

 

连接成功后，我们就可以通过MQTT服务器控制ESP32开发板了，此处需要先在Thoony IDE上运行我们之前的MQTT客户端代码。代码在第六节。

然后我们就可以通过手机的MQTT调试器进行“订阅”和“发布消息”了。

 

 

 

 

 

 发布后就可以手机控制开关灯了。

 

 然后在Thonny IDE中也会打印出相关操作的信息。

PS：大家无线WIIFI尽量使用2.4G的频率，5G的我没验证，且ESP32不支持5G的频率。

如有不明白的，请评论区留言哈。。。文章来源地址https://uudwc.com/A/r8Dv