Python模拟MQTT v3.1.1服务器

示例代码

import loggingimport asynciofrom hbMQtt.broker import Broker# 设置日志级别为DEBUGlogging.basicConfig(level=logging.DEBUG)# 创建MQtT服务器broker = Broker()# 启动MQTT服务器async def start_broker():    await broker.start()# 停止MQTT服务器async def stop_broker():    await broker.stop()# 主函数async def main():    # 启动MQTT服务器    await start_broker()    try:        # 保持主程序运行        while True:            await asyncio.sleep(1)    except KeyboardInterrupt:        # 捕捉键盘中断信号        pass    # 停止MQTT服务器    await stop_broker()# 运行主程序if __name__ == "__main__":    loop = asyncio.get_event_loop()    loop.run_until_complete(main())

运行结果

"D:\Program Files (x86)\python310\Python.exe" E:/文档/Projects/PyCharm/forklift/MQTT_SERVER.pyTraceback (most recent call last):  File "E:\文档\Projects\Pycharm\forklift\MQTT_SERVER.py", line 10, in <module>    from hbmqtt.broker import Broker  File "D:\Program Files (x86)\python310\lib\site-packages\hbmqtt\broker.py", line 15, in <module>    from hbmqtt.session import Session  File "D:\Program Files (x86)\Python310\lib\site-packages\hbmqtt\session.py", line 8, in <module>    from hbmqtt.mqtt.publish import PublishPacket  File "D:\Program Files (x86)\Python310\lib\site-packages\hbmqtt\mqtt\__init__.py", line 5, in <module>    from hbmqtt.mqtt.packet import (  File "D:\Program Files (x86)\Python310\lib\site-packages\hbmqtt\mqtt\packet.py", line 8, in <module>    from hbmqtt.adapters import ReaderAdapter, WriterAdapter  File "D:\Program Files (x86)\Python310\lib\site-packages\hbmqtt\adapters.py", line 6, in <module>    from websockets.protocol import WEBSocketCommonProtocolImportError: cannot import name 'WebSocketCommonProtocol' from 'websockets.protocol' (D:\Program Files (x86)\Python310\lib\site-packages\websockets\protocol.py)Process finished with exit code 1

cannot import name ‘WebSocketCommonProtocol’ from ‘websockets.protocol’

解决办法：
降低websockets版本

pip install websockets==8.1

新报错：

"D:\Program Files (x86)\Python385\python3.exe" E:/文档/Projects/Pycharm/forklift/MQTT_SERVER.pyTraceback (most recent call last):  File "D:\Program Files (x86)\Python385\lib\site-packages\hbmqtt\broker.py", line 185, in _build_listeners_config    listeners_config = broker_config['listeners']KeyError: 'listeners'During handling of the above exception, another exception occurred:Traceback (most recent call last):  File "E:/文档/Projects/Pycharm/forklift/MQTT_SERVER.py", line 17, in <module>    broker = Broker()  File "D:\Program Files (x86)\Python385\lib\site-packages\hbmqtt\broker.py", line 157, in __init__    self._build_listeners_config(self.config)  File "D:\Program Files (x86)\Python385\lib\site-packages\hbmqtt\broker.py", line 192, in _build_listeners_config    raise BrokerException("Listener config not found invalid: %s" % ke)hbmqtt.broker.BrokerException: Listener config not found invalid: 'listeners'Process finished with exit code 1

解决办法：
根据错误信息，看起来问题出在您的MQTT服务器配置中缺少了"listeners"配置项。您可以在实例化Broker对象时，提供一个有效的配置，包括"listeners"配置项。

config = {    "listeners": {        "default": {            "type": "tcp",            "bind": "localhost:1883",  # 监听本地1883端口            "max_connections": 10  # 最大连接数        }    },    "sys_interval": 10,    "topic-check": {        "enabled": False    }}broker = Broker(config)

参考：《mqtt 客户端和服务端搭建及基本使用详解》

下载客户端

打开客户端，新建连接

Github教程

创建教程

本地MQTT服务器

import osimport subprocess# 启动Mosquitto MQTT服务器def start_mosquitto_server():    command = "mosquitto"    process = subprocess.Popen(command, shell=True)    process.wait()# 停止Mosquitto MQTT服务器def stop_mosquitto_server():    command = "pkill mosquitto"    os.system(command)# 主函数def main():    try:        # 启动MQTT服务器        start_mosquitto_server()        # 保持主程序运行        while True:            pass    except KeyboardInterrupt:        # 捕捉键盘中断信号        pass        # 停止MQTT服务器        stop_mosquitto_server()# 运行主程序if __name__ == "__main__":    main()

系统找不到mosquitto命令，这可能是因为没有安装或正确配置Mosquitto MQTT服务器软件。

要解决这个问题，需要确保已经正确安装了Mosquitto MQTT服务器软件，并将其路径配置到系统的环境变量中。

安装Mosquitto MQTT服务器软件：

从Mosquitto官方网站（https://mosquitto.org/）下载并安装适合你操作系统的Mosquitto MQTT服务器软件。

配置系统环境变量：
将Mosquitto MQTT服务器软件的安装路径添加到系统的环境变量中，这样系统就能够找到mosquitto命令。

按照以下步骤进行配置：

打开系统的“高级系统设置”
在“系统变量”部分，找到名为“Path”的变量，选中它并点击“编辑”按钮
在编辑窗口中，点击“新建”按钮，并将Mosquitto MQTT服务器软件的安装路径添加进去
确认所有对话框并保存更改

重新运行代码：在完成以上步骤后，关闭并重新打开命令行窗口，然后再次运行你的代码，系统应该能够找到并启动mosquitto命令。



重启电脑后，依然读取不到mosquitto：

"D:\Program Files (x86)\Python310\python310.exe" E:/文档/Projects/Pycharm/forklift/MQTT_SERVER_0629_01.py'mosquitto' 不是内部或外部命令，也不是可运行的程序或批处理文件。

在任务栏看到了该服务，说明服务已经启动，通过客户端模拟，成功发送了数据

import paho.mqtt.client as mqtt# 定义回调函数，用于处理接收到的消息def on_message(client, userdata, msg):    print("Received message:", msg.topic, msg.payload.decode())# 创建MQTT客户端实例client = mqtt.Client()# 配置MQTT服务器地址和端口broker_address = "mqtt://localhost:1883"client.connect(broker_address)# 设置订阅回调函数client.on_message = on_message# 订阅主题topic = "0629tt"client.subscribe(topic)try:    # 启动MQTT客户端循环    client.loop_forever()except KeyboardInterrupt:    # 捕捉键盘中断信号    pass# 停止MQTT客户端循环和断开连接client.loop_stop()client.disconnect()

pip install paho-mqtt

"D:\Program Files (x86)\Python310\python310.exe" E:/文档/Projects/Pycharm/forklift/MQTT_SERVER_0629_01.pyTraceback (most recent call last):  File "E:\文档\Projects\Pycharm\forklift\MQTT_SERVER_0629_01.py", line 19, in <module>    client.connect(broker_address)  File "D:\Program Files (x86)\Python310\lib\site-packages\paho\mqtt\client.py", line 914, in connect    return self.reconnect()  File "D:\Program Files (x86)\Python310\lib\site-packages\paho\mqtt\client.py", line 1044, in reconnect    sock = self._create_socket_connection()  File "D:\Program Files (x86)\Python310\lib\site-packages\paho\mqtt\client.py", line 3685, in _create_socket_connection    return socket.create_connection(addr, timeout=self._connect_timeout, source_address=source)  File "D:\Program Files (x86)\Python310\lib\socket.py", line 824, in create_connection    for res in getaddrinfo(host, port, 0, SOCK_STREAM):  File "D:\Program Files (x86)\Python310\lib\socket.py", line 955, in getaddrinfo    for res in _socket.getaddrinfo(host, port, family, type, proto, flags):socket.gaierror: [Errno 11001] getaddrinfo failedProcess finished with exit code 1

解决办法

# 配置MQTT服务器地址和端口broker_address = "mqtt://localhost:1883"client.connect(broker_address)

修改为：

# 配置MQTT服务器地址和端口broker_address = "mqtt://127.0.0.1:1883"client.connect(broker_address)

依然无法访问

Try Except 打印出错误



参考文章

C:\windows\System32\drivers\etc

在使用 paho-mqtt 库时，broker_address 的格式应该是 host:port，而不是使用 mqtt:// 协议前缀。
修改代码中的 broker_address 如下：

broker_address = 'localhost'  # MQTT 服务器的主机名或 IP 地址port = 1883  # MQTT 服务器的端口号try:    client.connect(broker_address, port=port)except Exception as e:    print("connect error:", e)

源码：

import paho.mqtt.client as mqtt# 定义回调函数，用于处理接收到的消息def on_message(client, userdata, msg):    print("Received message:", msg.topic, msg.payload.decode())# 创建MQTT客户端实例client = mqtt.Client()# 配置MQTT服务器地址和端口broker_address = 'localhost'  # MQTT 服务器的主机名或 IP 地址port = 1883  # MQTT 服务器的端口号try:    client.connect(broker_address, port=port)except Exception as e:    print("connect error:", e)# 设置订阅回调函数client.on_message = on_message# 订阅主题topic = "0629tt"client.subscribe(topic)try:    # 启动MQTT客户端循环    client.loop_forever()except KeyboardInterrupt:    # 捕捉键盘中断信号    pass# 停止MQTT客户端循环和断开连接client.loop_stop()client.disconnect()

其他参考代码
打开 Mosquitto 的配置文件。在 Windows 系统上，
配置文件通常位于 Mosquitto 安装目录的 \mosquitto\mosquitto.conf 或 \mosquitto\mosquitto.conf 文件中。

找到 listener 部分的配置。如果找不到该部分，请在文件的末尾添加以下内容：

listener 1883bind_address 192.168.183.176

参考链接：《Mosquitto 安装指南》

C:\Program Files\mosquitto> mosquitto.exe -c mosquitto.conf -v

listener 1883allow_anonymous true

通讯测试结果：

success！！

订阅全部消息

可以将主题设置为通配符 #。通配符 # 表示匹配主题层级中的任意层级。

import paho.mqtt.client as mqtt# 定义回调函数，用于处理接收到的消息def on_message(client, userdata, msg):    print("Received message:", msg.topic, msg.payload.decode())# 创建MQTT客户端实例client = mqtt.Client()# 配置MQTT服务器地址和端口broker_address = 'localhost'  # MQTT 服务器的主机名或 IP 地址port = 1883  # MQTT 服务器的端口号try:    client.connect(broker_address, port=port)except Exception as e:    print("connect error:", e)# 设置订阅回调函数client.on_message = on_message# 订阅所有主题topic = "#"client.subscribe(topic)try:    # 启动MQTT客户端循环    client.loop_forever()except KeyboardInterrupt:    # 捕捉键盘中断信号    pass# 停止MQTT客户端循环和断开连接client.loop_stop()client.disconnect()

来源地址：https://blog.csdn.net/the_beginner/article/details/131440827