WebSockets MQTT 订阅

将 MQTT 客户端与 WebSocket 连接相结合。

WebSockets MQTT 订阅演示了如何设计自定义的QIODevice ，以将WebSocket连接与QMqttClient 结合使用。

注意：自 Qt 6.10起，Qt MQTT 已原生支持 WebSockets。保留此示例是为了演示如何实现自定义传输协议。

创建自定义 QIODevice

新的自定义设备WebSocketIODevice 必须是QIODevice 的子类：

class WebSocketIODevice : public QIODevice
{
    Q_OBJECT
public:
    WebSocketIODevice(QObject *parent = nullptr);

    bool isSequential() const override;
    qint64 bytesAvailable() const override;

    bool open(OpenMode mode) override;
    void close() override;

    qint64 readData(char *data, qint64 maxlen) override;
    qint64 writeData(const char *data, qint64 len) override;

    void setUrl(const QUrl &url);
    void setProtocol(const QByteArray &data);
Q_SIGNALS:
    void socketConnected();

public slots:
    void handleBinaryMessage(const QByteArray &msg);
    void onSocketConnected();

private:
    QByteArray m_protocol;
    QByteArray m_buffer;
    QWebSocket m_socket;
    QUrl m_url;
};

设计用于管理连接和订阅的类

WebSocketIODevice 将作为ClientSubscription 类的私有成员，与QMqttClient 和QMqttSubscription 并列：

private:
    QMqttClient m_client;
    QMqttSubscription *m_subscription;
    QUrl m_url;
    QString m_topic;
    WebSocketIODevice m_device;
    int m_version;

订阅和接收消息

主要逻辑实现于ClientSubscription 类的connectAndSubscribe() 方法中。在通过 WebSocket 初始化 MQTT 连接之前，您需要验证 WebSocket 是否已成功连接。MQTT 连接建立后，QMqttClient 即可订阅该主题。如果订阅成功，则可通过QMqttSubscription 从已订阅的主题接收消息，这些消息将由ClientSubscription 类的handleMessage() 方法进行处理。

voidClientSubscription::connectAndSubscribe()
{
    qCDebug(lcWebSocketMqtt)<< "正在连接到位于 " <<m_url<<" 的消息代理";

    m_device.setUrl(m_url);
    m_device.setUrl(m_url);

    m_device.setProtocol(m_version== 3 ? "mqttv3.1":"mqtt");

    connect(&m_device, &WebSocketIODevice::socketConnected, this, [this]() {
        qCDebug(lcWebSocketMqtt)<< "WebSocket 已连接，正在初始化 MQTT 连接。";

        m_client.setProtocolVersion(m_version== 3 ?QMqttClient::MQTT_3_1 : QMqttClient::MQTT_3_1_1);
        m_client.setTransport(&m_device,QMqttClient::IODevice);

        connect(&m_client, &QMqttClient::connected, this, [this]() {
            qCDebug(lcWebSocketMqtt)<< "MQTT 连接建立成功";

            m_subscription=m_client.subscribe(m_topic);
            if(!m_subscription) {
                qDebug() << "Failed to subscribe to " << m_topic;
               emiterrorOccured();
            }

            connect(m_subscription, &QMqttSubscription::stateChanged,
                    [](QMqttSubscription::SubscriptionState s) {
                qCDebug(lcWebSocketMqtt)<< "订阅状态已更改：" <<s;
            });

            connect(m_subscription, &QMqttSubscription::messageReceived,
                    [this](QMqttMessage msg) {
                handleMessage(msg.payload());
            });
        });

        m_client.connectToHost();
    });
    if(!m_device.open(QIODevice::ReadWrite))
        qDebug() << "Could not open socket device";
}

文件：

© 2026 The Qt Company Ltd. Documentation contributions included herein are the copyrights of their respective owners. The documentation provided herein is licensed under the terms of the GNU Free Documentation License version 1.3 as published by the Free Software Foundation. Qt and respective logos are trademarks of The Qt Company Ltd. in Finland and/or other countries worldwide. All other trademarks are property of their respective owners.