【案例共创】在开发者空间快速开发 MQTT 客户端实现硬件仿真上云

2025-12-15
贵州
本文字数：6662 字
阅读完需：约 22 分钟

最新案例动态，请查阅【案例共创】在开发者空间快速开发MQTT客户端实现硬件仿真上云。小伙伴们快来领取华为开发者空间进行实操吧！

本案例由开发者：DS 小龙哥提供

1 概述

1.1 背景介绍

随着物联网技术的不断发展，越来越多的设备和应用依赖于实时数据交换和远程控制。在物联网生态系统中，设备与云平台之间的通信是核心环节之一，然而对于许多开发者来说，进行这种设备与云平台之间的通信往往涉及到硬件的配置与调试，这对于一些不熟悉硬件的开发者，尤其是那些处于软件开发领域的人员，可能是一大挑战。传统的物联网开发往往需要开发者拥有一定的硬件基础，或者至少具备与硬件设备进行调试和交互的能力，这使得一些开发者在没有硬件设备的情况下，难以快速上手和测试物联网应用，开发者迫切需要一种能够模拟硬件设备并与云平台进行交互的工具。

本案例通过开发一款基于 MQTT 协议的客户端调试助手，旨在为开发者提供一个简单易用的工具，模拟硬件设备与云平台的通信交互。这款工具通过软件模拟了物联网设备的行为，支持主题的订阅与发布，能够与华为云物联网平台（IoTDA）进行实时通信。对于不熟悉硬件的开发者，或者暂时没有硬件设备的开发者而言，这款调试助手可以让他们在没有物理硬件的前提下，体验完整的物联网设备上云过程。开发者可以通过该工具快速了解设备如何连接云平台，如何进行数据传输，并学习 MQTT 协议的基本操作。

1.2 适用对象

企业
个人开发者
高校学生

1.3 案例时间

本案例总时长预计 60 分钟。

1.4 案例流程

说明：

登录开发者空间，配置开发环境；
编辑 MQTT 客户端源码；
CodeArts IDE 运行 MQTT 客户端源码文件；
MQTT 客户端实现与 MQTT 服务端通信，实现连接 MQTT 服务器、发布和订阅消息；
MQTT 服务器与华为云 IoTA 通信，像注册的设备发布和订阅消息。

1.5 资源总览

本案例预计花费总计 0 元。

2 开发者空间开发环境准备

本案例中，实现 MQTT 客户端与云端注册的设备进行交互，需要开通 IoTA 服务以及安装开发客户端所需依赖库。

2.1 配置云主机

登录开发者空间，登录后页面如下：

点击“配置云主机”，在弹出的对话框中进行云主机配置。

按如规格下配置云主机：
云主机名称：默认/自定义
CPU 架构：X86
规格：2 vCPUs 4 GB
操作系统：Ubuntu
系统镜像：公共镜像 Ubuntu 22.04 server 64bit (xfce4 desktop)
工具：Python 工具集（CodeArts IDE+ Python +Git）

确认以上配置无误，点击“安装”，进行云主机操作系统安装。

安装完毕之后，点击“进入桌面”。

环境准备中，大约需要 3-5 分钟，请您耐心等待…

进入桌面后的默认效果如下：

点击左下角的“所有应用程序”->“开发”->“CodeArts IDE for Python”，打开 IDE。

CodeArts IDE for Python 启动后，在弹框界面，选择“新建工程”。

在新建工程页面，自定义输入工程名称，点击“创建”。

在 CodeArts IDE for Python 中，在新建的工程文件目录中，选择“venv/lib/python3.10/site-packages”路径下的任一文件，鼠标右键后，选择“打开所在文件夹”。

复制被打开的文件夹路径。

在 CodeArts IDE for Python 中，点击下方的“终端”，输入以下命令后回车，安装 paho-mqtt 库（paho-mqtt 是一个提供 MQTT 协议功能的 Python 库，通过这个库，开发者可以快速实现 MQTT 客户端的功能，包括连接到 MQTT 代理服务器、发布消息到主题、订阅感兴趣的主题以及接收并处理消息。）：

pip install paho-mqtt --target={package-path}

复制代码

其中{package-path}用上面复制的文件夹路径替换。

按上面的方式执行以下命令，安装 PyQt5 库（PyQt5 是基于 paho-mqtt 库，实现 MQTT 通信。）：

pip install PyQt5 --target={package-path}

复制代码

其中{package-path}也用上面复制的文件夹路径替换。

到此，云主机的开发环境已经配置完成。

2.2 开通 IoTA 服务

登录设备接入IoTA服务控制台，点击“开通免费单元”。

实例配置保持默认，点击“立即创建”按钮。

需要等待标准版实例创建完成。

创建 IoT 设备

2.3 创建产品

实例创建完成之后，点击实例名称，进入实例。点击左侧“产品”菜单栏，点击“创建产品”按钮。

在“创建产品”弹窗中，自定义填写产品名称，设备类型选择“自定义类型”，自定义填写设备类型（例：dev），点击“确定”，完成产品创建。

在“创建产品成功”提示窗中点击“查看详情”。

在产品详情页面，点击“自定义模型”，在“添加服务”弹窗中，填写服务 ID（例：stm32），点击“确定”。

说明：模型就是存放设备上传到云平台的数据，你可以根据自己的产品进行创建。

在新增的服务中，点击“新增属性”，在“新增属性”弹窗中，填写属性名称，点击“确定”。设备属性‌是指与物联网设备相关的各种参数和设置，这些属性通常以键值对的形式存在，用于描述设备的各种特征和行为。

点击左上角“<”回到上一级页面。

2.4 添加设备

产品是属于上层的抽象模型，接下来在产品模型下添加实际的设备。添加的设备最终需要与真实的设备关联在一起，完成数据交互。

在左侧菜单栏选择“设备->所有设备”，点击“注册设备”。

在“单设备注册”弹窗中，选择所属资源空间，所属产品选择步骤 3.1 中创建的产品，自定义输入设备标识码（例：dev1）、设备名称和秘钥，点击“确定”。

在“设备创建成功”提示窗中，点击“保存并关闭”。可以看到，刚刚注册的设备处于“未激活”状态，待真实设备接入平台才会变成“在线”状态。

2.5 生成 MQTT 三元组

华为云提供了一个在线工具，用来生成 MQTT 鉴权三元组。

打开这个MQTT ClientId生成工具，DeviceId 填入刚刚注册设备的设备 ID，DeviceSecret 填入步骤 3.2 中注册设备时设置的秘钥，点击“Generate”，就可以得到 MQTT 的登录信息了。

图形化界面开发 MQTT 客户端

下面我们开发完成 MQTT 客户端调试助手，模拟真实设备接入 IoTA 平台，整体开发，基于 paho-mqtt 库来实现以下功能：

连接到 MQTT 服务器：通过提供的 IP、端口、客户端 ID、用户名和密码连接到 MQTT 服务器。

订阅主题：从用户输入的订阅主题中接收消息。

发布消息：发布主题消息到指定的发布主题。

每个按钮添加相应的功能：Connect、订阅和发布。在此基础上，日志框将显示与 MQTT 连接和消息传输相关的调试信息。

在云主机的 CodeArts IDE for Python 中，点击“文件”->“新建”->“文件”。

点击“文件”->“保存”。

输入文件名称为“MQTT.py”，点击“保存”。

在 MQTT.py 文件中输入以下代码（复制文档中 python 代码时，可能会导致格式错误，可以点击下载获取 MQTT.py 文件内容！），用于实现 MQTT 客户端：

import sysimport jsonimport paho.mqtt.client as mqtt # 导入 paho-mqtt 库from PyQt5.QtCore import Qtfrom PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QFormLayout, QLabel, QLineEdit, QSpinBox, QPushButton, QGridLayout, QGroupBox, QPlainTextEdit, QSpacerItem, QSizePolicy, QMenuBar, QStatusBarimport timeclass MQTTClientDebugger(QMainWindow):def __init__(self):super().__init__()self.connected = Falseself.setWindowTitle("MQTT 客户端调试助手")self.setGeometry(100, 100, 1019, 772) # 设置窗口大小self.centralWidget = QWidget(self)self.setCentralWidget(self.centralWidget)self.client = None # MQTT 客户端实例# 主布局self.mainLayout = QVBoxLayout(self.centralWidget)# 连接设置布局self.connectionLayout = QHBoxLayout()self.host = "117.78.5.125"self.clientId = ""self.username = ""self.passWord = ""self.formLayout = QFormLayout()self.hostLineEdit = QLineEdit()self.hostLineEdit.setText(self.host) \# 默认服务器IP地址self.formLayout.addRow(QLabel("服务器域名或者IP地址:"), self.hostLineEdit)# 端口号self.spinBoxPort = QSpinBox()self.spinBoxPort.setMaximum(99999)self.spinBoxPort.setValue(1883) # 默认端口号self.formLayout.addRow(QLabel("服务器端口:"), self.spinBoxPort)self.clientIdLineEdit = QLineEdit()self.clientIdLineEdit.setText(self.clientId) #客户端IDself.formLayout.addRow(QLabel("ClientId"), self.clientIdLineEdit)self.usernameLineEdit = QLineEdit()self.usernameLineEdit.setText(self.username) # 设备用户名self.formLayout.addRow(QLabel("Username"), self.usernameLineEdit)self.passwordLineEdit = QLineEdit()self.passwordLineEdit.setText(self.passWord) # 默认密码self.formLayout.addRow(QLabel("Password"), self.passwordLineEdit)self.connectionLayout.addLayout(self.formLayout)# 连接按钮self.connectButton = QPushButton("Connect")self.connectButton.clicked.connect(self.connect_to_server)self.connectionLayout.addWidget(self.connectButton)self.mainLayout.addLayout(self.connectionLayout)# MQTT 主题和消息布局self.gridLayout = QGridLayout()self.gridLayout.addWidget(QLabel("订阅主题:"), 0, 0)self.subscribeTopicLineEdit = QLineEdit("\$oc/devices/"+ self.username +"/sys/messages/down")self.gridLayout.addWidget(self.subscribeTopicLineEdit, 0, 1)self.gridLayout.addWidget(QPushButton("订阅"), 0, 2)self.gridLayout.addWidget(QLabel("发布主题:"), 1, 0)self.publishTopicLineEdit = QLineEdit("\$oc/devices/"+ self.username +"/sys/properties/report")self.gridLayout.addWidget(self.publishTopicLineEdit, 1, 1)self.gridLayout.addWidget(QLabel("主题消息:"), 2, 0)self.messageLineEdit = QLineEdit('{"services": [{"service_id": "stm32","properties":{"DHT11_T":18.1,"DHT11_H":16.2,"SOIL":12.4,"BH1750":124.5,"MOTOR_SW":1,"SOIL_MAX":30,"run_mode":1}}]}')self.gridLayout.addWidget(self.messageLineEdit, 2, 1)self.publishButton = QPushButton("发布")self.publishButton.clicked.connect(self.publish_message)self.gridLayout.addWidget(self.publishButton, 2, 2)self.mainLayout.addLayout(self.gridLayout)# 日志区域self.logGroupBox = QGroupBox("日志消息:")self.logLayout = QHBoxLayout()self.logTextEdit = QPlainTextEdit()self.logTextEdit.setReadOnly(True)self.logLayout.addWidget(self.logTextEdit)self.logGroupBox.setLayout(self.logLayout)self.mainLayout.addWidget(self.logGroupBox)# 底部按钮布局self.bottomLayout = QHBoxLayout()self.testButton = QPushButton("测试按钮(一键填充MQTT信息)")self.clearButton = QPushButton("一键清除MQTT信息")self.clearLogButton = QPushButton("清除日志消息")self.viewTutorialButton = QPushButton("【查看物联网项目开发教程】")self.quitButton = QPushButton("退出软件")self.testButton.clicked.connect(self.fillMQTTInfo)self.clearButton.clicked.connect(self.clearMQTTInfo)self.clearLogButton.clicked.connect(self.clear_logs)self.bottomLayout.addWidget(self.testButton)self.bottomLayout.addWidget(self.clearButton)self.bottomLayout.addWidget(self.clearLogButton)self.bottomLayout.addWidget(self.viewTutorialButton)self.bottomLayout.addWidget(self.quitButton)self.mainLayout.addLayout(self.bottomLayout)# 菜单栏self.menuBar = self.menuBar()self.fileMenu = self.menuBar.addMenu("File")quitAction = self.fileMenu.addAction("Quit")quitAction.triggered.connect(self.close)# 状态栏self.statusBar = QStatusBar()self.setStatusBar(self.statusBar)# 连接到MQTT服务器def connect_to_server(self):host = self.hostLineEdit.text()port = self.spinBoxPort.value()client_id = self.clientIdLineEdit.text()username = self.usernameLineEdit.text()password = self.passwordLineEdit.text()try:self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1,client_id)self.client.username_pw_set(username, password) # 设置用户名和密码# 设置连接成功、消息接收、连接丢失等回调函数self.client.on_connect = self.on_connectself.client.on_message = self.on_messageself.client.on_disconnect = self.on_disconnect# 连接到服务器self.client.connect(host, port, 60)# 启动 MQTT 客户端self.client.loop_start()while not self.connected and not self._stop_event.is_set():time.sleep(0.1)self.connected = Trueexcept Exception as e:return f"{e}"def on_connect(self, client, userdata, flags, rc):"""当连接到MQTT服务器时调用"""self.log(f"连接成功，返回码：{rc}")# 连接成功后订阅主题self.subscribeTopicLineEdit = QLineEdit("\$oc/devices/"+ self.usernameLineEdit.text() +"/sys/messages/down")subscribe_topic = self.subscribeTopicLineEdit.text()try:client.subscribe(subscribe_topic)except Exception as e:self.log(f"订阅失败：{e}")def on_message(self, client, userdata, msg):"""当接收到MQTT消息时调用"""self.log(f"接收到消息：{msg.topic} {msg.payload.decode()}")def on_disconnect(self, client, userdata, rc):"""当断开连接时调用"""self.log(f"MQTT服务器断开连接，返回码：{rc}")# 发布消息def publish_message(self):self.publishTopicLineEdit = QLineEdit("\$oc/devices/"+ self.usernameLineEdit.text() +"/sys/properties/report")topic = self.publishTopicLineEdit.text()message = self.messageLineEdit.text()if self.client:self.client.publish(topic, message)self.log(f"发布消息：{topic} {message}")# 发送心跳包def send_heartbeat(self):if self.client:self.client.ping()self.log("发送心跳包")# 日志输出def log(self, message):"""向日志框输出信息"""self.logTextEdit.appendPlainText(message)# 填充MQTT信息def fillMQTTInfo(self):self.hostLineEdit.setText(self.host)self.spinBoxPort.setValue(1883)self.clientIdLineEdit.setText(self.clientId)self.usernameLineEdit.setText(self.username)self.passwordLineEdit.setText(self.passWord)self.subscribeTopicLineEdit.setText("\$oc/devices/"+ self.username +"/sys/messages/down")self.publishTopicLineEdit.setText("\$oc/devices/"+ self.username +"/sys/properties/report")self.messageLineEdit.setText('{"services": [{"service_id": "stm32","properties":{"DHT11_T":18.1,"DHT11_H":16.2,"SOIL":12.4,"BH1750":124.5,"MOTOR_SW":1,"SOIL_MAX":30,"run_mode":1}}]}')# 清除MQTT信息def clearMQTTInfo(self):self.hostLineEdit.clear()self.spinBoxPort.clear()self.clientIdLineEdit.clear()self.usernameLineEdit.clear()self.passwordLineEdit.clear()self.subscribeTopicLineEdit.clear()self.publishTopicLineEdit.clear()# 清除日志信息def clear_logs(self):self.logTextEdit.clear()if \__name_\_ == '__main__':app = QApplication(sys.argv)mainWin = MQTTClientDebugger()mainWin.show()sys.exit(app.exec_())