C++/Qt + Mqtt协议 实现客户端订阅/发布功能

目录

一、UI 构建（setupUI）

二、信号槽绑定（setupConnections）

三、连接/断开 Broker（onConnectClicked）

四、 订阅主题（onSubscribeClicked）

五、取消订阅（onUnsubscribeClicked）

六、发布普通消息（onPublishClicked）

七、发布保留消息（onPublishRetainClicked）

八、连接状态变化处理（onStateChanged）

九、连接成功处理（onConnected）

十、断开连接处理（onDisconnected）

十一、 接收消息处理（onMessageReceived）

十二、消息发送确认（onMessageSent）

十三、错误处理（onErrorChanged）

一、UI 构建（setupUI）

void MainWindow::setupUI() { // --- 中央部件和主布局 --- QWidget *central = new QWidget(this); setCentralWidget(central); QVBoxLayout *mainLayout = new QVBoxLayout(central); // ==================== 1. 连接配置区域 ==================== m_groupConnect = new QGroupBox(tr("连接配置"), this); QFormLayout *formLayout = new QFormLayout(m_groupConnect); // Broker 地址（默认 localhost） m_editHost = new QLineEdit("127.0.0.1", this); formLayout->addRow(tr("Broker 地址:"), m_editHost); // 端口号（MQTT 默认 1883） m_spinPort = new QSpinBox(this); m_spinPort->setRange(1, 65535); m_spinPort->setValue(1883); formLayout->addRow(tr("端口:"), m_spinPort); // 客户端 ID（MQTT 协议要求每个客户端有唯一 ID） // QUuid 生成一个全局唯一标识符，避免多客户端冲突 m_editClientId = new QLineEdit(QUuid::createUuid().toString(QUuid::WithoutBraces), this); formLayout->addRow(tr("客户端 ID:"), m_editClientId); // 用户名和密码（可选，用于 Broker 认证） m_editUsername = new QLineEdit(this); m_editUsername->setPlaceholderText("可选"); formLayout->addRow(tr("用户名:"), m_editUsername); m_editPassword = new QLineEdit(this); m_editPassword->setPlaceholderText("可选"); m_editPassword->setEchoMode(QLineEdit::Password); formLayout->addRow(tr("密码:"), m_editPassword); // QoS 等级选择 // QoS 0: 最多一次（发完不管，可能丢） // QoS 1: 至少一次（保证收到，可能重复） // QoS 2: 恰好一次（保证收到且不重复，开销最大） m_comboQos = new QComboBox(this); m_comboQos->addItem("QoS 0 - 最多一次", 0); m_comboQos->addItem("QoS 1 - 至少一次", 1); m_comboQos->addItem("QoS 2 - 恰好一次", 2); formLayout->addRow(tr("QoS 等级:"), m_comboQos); // 连接/断开按钮 + 状态标签 QHBoxLayout *connBtnLayout = new QHBoxLayout(); m_btnConnect = new QPushButton(tr("连接"), this); m_btnConnect->setMinimumWidth(100); connBtnLayout->addWidget(m_btnConnect); m_labelStatus = new QLabel(tr("状态: 未连接"), this); m_labelStatus->setStyleSheet("color: red; font-weight: bold;"); connBtnLayout->addWidget(m_labelStatus); connBtnLayout->addStretch(); formLayout->addRow(connBtnLayout); mainLayout->addWidget(m_groupConnect); // ==================== 2. 订阅区域 ==================== m_groupSubscribe = new QGroupBox(tr("订阅主题"), this); QHBoxLayout *subLayout = new QHBoxLayout(m_groupSubscribe); m_editSubTopic = new QLineEdit("test/topic", this); m_editSubTopic->setPlaceholderText("输入主题名，如 sensor/temp"); subLayout->addWidget(m_editSubTopic); m_btnSubscribe = new QPushButton(tr("订阅"), this); subLayout->addWidget(m_btnSubscribe); m_btnUnsubscribe = new QPushButton(tr("取消订阅"), this); subLayout->addWidget(m_btnUnsubscribe); mainLayout->addWidget(m_groupSubscribe); // ==================== 3. 发布区域 ==================== m_groupPublish = new QGroupBox(tr("发布消息"), this); QHBoxLayout *pubLayout = new QHBoxLayout(m_groupPublish); m_editPubTopic = new QLineEdit("test/topic", this); m_editPubTopic->setPlaceholderText("发布主题"); pubLayout->addWidget(m_editPubTopic, 2); m_editMessage = new QLineEdit("Hello MQTT", this); m_editMessage->setPlaceholderText("消息内容"); pubLayout->addWidget(m_editMessage, 3); m_btnPublish = new QPushButton(tr("发布"), this); pubLayout->addWidget(m_btnPublish); m_btnPubRetain = new QPushButton(tr("发布(保留)"), this); m_btnPubRetain->setToolTip("保留消息：Broker 会保存最后一条，新订阅者立即收到"); pubLayout->addWidget(m_btnPubRetain); mainLayout->addWidget(m_groupPublish); // ==================== 4. 消息日志区域 ==================== QGroupBox *logGroup = new QGroupBox(tr("消息日志"), this); QVBoxLayout *logLayout = new QVBoxLayout(logGroup); m_logText = new QTextEdit(this); m_logText->setReadOnly(true); // 只读，不允许用户编辑 logLayout->addWidget(m_logText); mainLayout->addWidget(logGroup, 1); // 日志区域占剩余空间 // 窗口基本设置 setWindowTitle(tr("MQTT 学习 Demo - QtMqtt")); resize(700, 650); }

二、信号槽绑定（setupConnections）

void MainWindow::setupConnections() { // --- 按钮点击信号 --- connect(m_btnConnect, &QPushButton::clicked, this, &MainWindow::onConnectClicked); connect(m_btnSubscribe, &QPushButton::clicked, this, &MainWindow::onSubscribeClicked); //订阅 connect(m_btnUnsubscribe, &QPushButton::clicked, this, &MainWindow::onUnsubscribeClicked); //取消订阅 connect(m_btnPublish, &QPushButton::clicked, this, &MainWindow::onPublishClicked); //发布普通消息 connect(m_btnPubRetain, &QPushButton::clicked, this, &MainWindow::onPublishRetainClicked); // 发布保留消息 // --- QMqttClient 核心信号（重点学习这些） --- // 1. stateChanged: 连接状态变化时触发 // ClientState 枚举值: Disconnected=0, Connecting=1, Connected=2 connect(m_client, &QMqttClient::stateChanged, this, &MainWindow::onStateChanged); // 2. connected: 成功连接到 Broker 时触发 connect(m_client, &QMqttClient::connected, this, &MainWindow::onConnected); // 3. disconnected: 与 Broker 断开连接时触发 connect(m_client, &QMqttClient::disconnected, this, &MainWindow::onDisconnected); // 4. messageReceived: 收到订阅主题的消息时触发 // 参数: QByteArray 消息内容, QMqttTopicName 主题名 connect(m_client, &QMqttClient::messageReceived, this, &MainWindow::onMessageReceived); // 5. messageSent: 消息成功发布到 Broker 时触发 // 参数: qint32 消息 ID connect(m_client, &QMqttClient::messageSent, this, &MainWindow::onMessageSent); // 6. errorChanged: 发生错误时触发 connect(m_client, &QMqttClient::errorChanged, this, &MainWindow::onErrorChanged); }

三、连接/断开 Broker（onConnectClicked）

void MainWindow::onConnectClicked() { if (m_client->state() == QMqttClient::Connected) { // 已连接 → 执行断开 m_client->disconnectFromHost(); return; } // --- 配置连接参数 --- m_client->setHostname(m_editHost->text()); m_client->setPort(m_spinPort->value()); m_client->setClientId(m_editClientId->text()); // 设置认证信息（如果填了用户名和密码） if (!m_editUsername->text().isEmpty()) { m_client->setUsername(m_editUsername->text()); } if (!m_editPassword->text().isEmpty()) { m_client->setPassword(m_editPassword->text()); } // 设置 Keep Alive（心跳间隔，单位：秒） // 客户端必须在此时间内至少发送一次消息或 PING 请求，否则 Broker 会认为客户端已死并断开连接 // 默认 60 秒，设为 30 秒更保险 m_client->setKeepAlive(30); /* * 什么是遗嘱消息？ 遗嘱消息是 MQTT 客户端在连接 Broker 时提前“立下遗嘱”：指定一个主题和一条消息。如果该客户端异常断开（例如网络故障、电源掉电、程序崩溃等，而不是主动发送 DISCONNECT 报文）， Broker 就会自动将这条遗嘱消息发布到指定的主题上，通知其他订阅者。 类比：一个人（客户端）在进入危险环境前，提前写好遗书（遗嘱消息）交给公证人（Broker），并说明如果自己意外死亡（异常断开），就把遗书公布给指定的人（订阅者）。 */ // 配置遗嘱消息（Last Will） // 作用：当客户端异常断开时，Broker 自动发布这条消息通知其他客户端 m_client->setWillTopic("client/status");//设置遗嘱消息的发布主题。当客户端异常断开时，Broker 会向 client/status 这个主题发布一条消息。 m_client->setWillMessage("客户端离线"); m_client->setWillQoS(1); m_client->setWillRetain(true); // 标记为保留消息。Broker 会永久保存这条消息，后续新订阅 client/status 主题的客户端立即收到这条离线通知，而不需要等待下一次事件发生。 appendLog("正在连接到 Broker " + m_editHost->text() + ":" + QString::number(m_spinPort->value()) + " ..."); // 发起连接（异步操作，结果通过 connected/disconnected 信号通知） m_client->connectToHost(); }

四、 订阅主题（onSubscribeClicked）

void MainWindow::onSubscribeClicked() { QString topic = m_editSubTopic->text().trimmed(); if (topic.isEmpty()) { QMessageBox::warning(this, tr("警告"), tr("请输入要订阅的主题名")); return; } // 获取用户选择的 QoS 等级 int qos = m_comboQos->currentData().toInt(); // QMqttClient::subscribe() 返回 QMqttSubscription 对象 // 可通过它监控订阅状态（如是否成功、QoS 协商结果等） /*MQTT 主题过滤器 可以包含通配符： +：单层通配符，匹配任意一个层级。 #：多层通配符，匹配任意多个层级（必须放在最后）。 当你构造 QMqttTopicFilter(topic) 时，如果传入的 topic 不包含通配符，它就是一个精确的主题过滤器，只匹配该主题本身；如果包含通配符，它会匹配所有符合条件的主题。*/ QMqttSubscription *sub = m_client->subscribe(QMqttTopicFilter(topic), qos); // QMqttClient::subscribe()订阅方法，返回订阅对象 // 可选：监听订阅状态变化 connect(sub, &QMqttSubscription::stateChanged, this, [this, topic](QMqttSubscription::SubscriptionState state) { if (state == QMqttSubscription::Subscribed) { appendLog("✓ 订阅成功: " + topic); } else if (state == QMqttSubscription::Unsubscribed) { appendLog("✗ 已取消订阅: " + topic); } }); appendLog("正在订阅主题: " + topic + " (QoS " + QString::number(qos) + ")"); }

五、取消订阅（onUnsubscribeClicked）

void MainWindow::onUnsubscribeClicked() { QString topic = m_editSubTopic->text().trimmed(); if (topic.isEmpty()) { QMessageBox::warning(this, tr("警告"), tr("请输入要取消订阅的主题名")); return; } // QMqttClient::unsubscribe() 取消对指定主题的订阅 m_client->unsubscribe(QMqttTopicFilter(topic)); appendLog("正在取消订阅: " + topic); }

六、发布普通消息（onPublishClicked）

void MainWindow::onPublishClicked() { QString topic = m_editPubTopic->text().trimmed(); QString message = m_editMessage->text().trimmed(); if (topic.isEmpty() || message.isEmpty()) { QMessageBox::warning(this, tr("警告"), tr("请输入主题和消息内容")); return; } int qos = m_comboQos->currentData().toInt(); // QMqttClient::publish() 发布消息 // 参数: 主题名, 消息内容, QoS 等级, 是否保留（默认false） // 返回值: qint32 消息 ID，可用于跟踪发布状态 qint32 msgId = m_client->publish( QMqttTopicName(topic), // 主题 message.toUtf8(), // 消息体（转为 UTF-8 字节） qos // QoS 等级 ); appendLog("发布消息 -> 主题: " + topic + ", QoS: " + QString::number(qos) + ", ID: " + QString::number(msgId)); }

七、发布保留消息（onPublishRetainClicked）

void MainWindow::onPublishRetainClicked() { QString topic = m_editPubTopic->text().trimmed(); QString message = m_editMessage->text().trimmed(); if (topic.isEmpty() || message.isEmpty()) { QMessageBox::warning(this, tr("警告"), tr("请输入主题和消息内容")); return; } int qos = m_comboQos->currentData().toInt(); // 保留消息：Broker 会保存这条消息，后续订阅该主题的新客户端会立即收到 // 典型用途：设备状态、配置信息等"最新值" qint32 msgId = m_client->publish( QMqttTopicName(topic), message.toUtf8(), qos, true // retain = true，标记为保留消息 ); appendLog("发布保留消息 -> 主题: " + topic + ", ID: " + QString::number(msgId)); }

八、连接状态变化处理（onStateChanged）

void MainWindow::onStateChanged(QMqttClient::ClientState state) { // 根据当前状态更新 UI 显示 QString stateText; switch (state) { case QMqttClient::Disconnected: stateText = "未连接"; break; case QMqttClient::Connecting: stateText = "正在连接..."; break; case QMqttClient::Connected: stateText = "已连接"; break; } m_labelStatus->setText("状态: " + stateText); // 连接中显示黄色，已连接显示绿色，未连接显示红色 if (state == QMqttClient::Connected) { m_labelStatus->setStyleSheet("color: green; font-weight: bold;"); } else if (state == QMqttClient::Connecting) { m_labelStatus->setStyleSheet("color: orange; font-weight: bold;"); } else { m_labelStatus->setStyleSheet("color: red; font-weight: bold;"); } }

九、连接成功处理（onConnected）

void MainWindow::onConnected() { appendLog("===== 已连接到 Broker ====="); m_btnConnect->setText("断开"); // 连接成功后，可以发布一条上线通知 m_client->publish(QMqttTopicName("client/status"), "客户端在线", 1, true); }

十、断开连接处理（onDisconnected）

void MainWindow::onDisconnected() { appendLog("===== 与 Broker 断开连接 ====="); m_btnConnect->setText("连接"); }

十一、 接收消息处理（onMessageReceived）

void MainWindow::onMessageReceived(const QByteArray &message, const QMqttTopicName &topic) { // 收到消息时的核心处理逻辑 // topic.name() → 获取主题名字符串 // message → 获取消息内容（QByteArray，可转 QString） QString timeStr = QDateTime::currentDateTime().toString("HH:mm:ss"); appendLog("[" + timeStr + "] 收到消息 ← 主题: " + topic.name() + " | 内容: " + QString::fromUtf8(message)); }

十二、消息发送确认（onMessageSent）

void MainWindow::onMessageSent(qint32 id) { // 消息成功发布到 Broker 后触发 // 注意：QoS 0 时可能不会触发此信号（因为不需要确认） appendLog("消息发送确认 ✓ (ID: " + QString::number(id) + ")"); }

十三、错误处理（onErrorChanged）

void MainWindow::onErrorChanged(QMqttClient::ClientError error) { // 错误处理：根据错误码判断问题类型 QString errorText; switch (error) { case QMqttClient::NoError: return; // 无错误，忽略 // --- 协议层错误（MQTT 协议返回） --- case QMqttClient::InvalidProtocolVersion: errorText = "协议版本不支持"; break; case QMqttClient::IdRejected: errorText = "客户端 ID 被拒绝（可能太长或格式不对）"; break; case QMqttClient::ServerUnavailable: errorText = "Broker 不可用"; break; case QMqttClient::BadUsernameOrPassword: errorText = "用户名或密码错误"; break; case QMqttClient::NotAuthorized: errorText = "认证失败，无权限"; break; // --- 传输层错误（网络问题） --- case QMqttClient::TransportInvalid: errorText = "传输层错误（网络断开/连接失败）"; break; case QMqttClient::ProtocolViolation: errorText = "协议违规（消息格式不正确）"; break; case QMqttClient::UnknownError: errorText = "未知错误"; break; case QMqttClient::Mqtt5SpecificError: errorText = "MQTT 5.0 特有错误"; break; default: errorText = "未知错误码: " + QString::number(error); break; } appendLog("⚠ 错误: " + errorText); }