MochiuWiki : SUSE, EC, PCB

「Qtの基礎 - MQTT」の版間の差分

📢 Webサイト閉鎖と移転のお知らせ
このWebサイトは2026年9月に閉鎖いたします。
新しい記事は移転先で追加しております。（旧サイトでは記事を追加しておりません）

← 古い編集
ビジュアルウィキテキスト
 
(同じ利用者による、間の6版が非表示)
21行目: 21行目:
<br>
<br>
また、大量のメッセージを送受信する場合は、非同期処理を活用してUIのブロッキングを防ぐことが重要である。<br>
また、大量のメッセージを送受信する場合は、非同期処理を活用してUIのブロッキングを防ぐことが重要である。<br>
<br><br>
== QtMqttモジュールのインストール ==
==== パッケージ管理システムからインストール ====
# RHEL
sudo dnf install qt6-qtmqtt-devel
# RHEL
sudo zypper install qt6-mqtt-devel
<br>
==== ソースコードからインストール ====
QtMqttモジュールのソースコードをダウンロードする。<br>
git clone --depth 1 git://code.qt.io/qt/qtmqtt.git -b <Qtのブランチ  例: 6.5.3>
cd qtmqtt
<br>
QtMqttモジュールをビルドおよびインストールする。<br>
<syntaxhighlight lang="sh">
mkdir build && cd build
# QtオンラインインストーラからQtライブラリをインストールしている場合
## Qt 5の場合
/<Qtのインストールディレクトリ>/<バージョン>/gcc_64/bin/qmake ../qtmqtt.pro
make -j $(nproc)
make install
## Qt 6 の場合
/<Qtのインストールディレクトリ>/<バージョン>/gcc_64/bin/qt-cmake \
    -DCMAKE_BUILD_TYPE=Release \
    -DCMAKE_C_COMPILER=<GCC 8以降のGCCのパス>  \
    -DCMAKE_CXX_COMPILER=<GCC 8以降のG++のパス> \
    ..
make -j $(nproc)
make install
# パッケージ管理システムからQtライブラリをインストールしている場合
## Qt 5の場合
qmake ../qtmqtt.pro
make -j $(nproc)
make install
## Qt 6 の場合
cmake -DCMAKE_BUILD_TYPE=Release \
      -DCMAKE_INSTALL_PREFIX="<QtMqttモジュールのインストールディレクトリ>" \
      -DCMAKE_C_COMPILER=<GCC 8以降のGCCのパス>  \
      -DCMAKE_CXX_COMPILER=<GCC 8以降のG++のパス> \
      ..
make -j $(nproc)
make install
</syntaxhighlight>
<br><br>
== MQTT関連のクラスとメソッド ==
==== QMqttClientクラスのメソッド ====
基本的な接続に関する設定を以下に示す。<br>
<br>
<syntaxhighlight lang="c++">
QMqttClient client;
// MQTTブローカーのホスト名の指定
client->setHostname("localhost");
// ポート番号の指定
client->setPort(1883);
// ユーザ名の指定 (認証が必要な場合)
client->setUsername("<ユーザ名>");
// パスワード (認証が必要な場合)
client->setPassword("<パスワード>");
// MQTTブローカーの接続
client->connectToHost();
// MQTTブローカーの切断
client->disconnectFromHost();
// 現在の接続状態の取得
// 状態の種類:
// -> QMqttClient::Disconnected
// -> QMqttClient::Connecting
// -> QMqttClient::Connected
QMqttClient::ClientState state = client->state();
</syntaxhighlight>
<br>
MQTTメッセージの保持 (Retain) フラグとは、MQTTブローカーがそのトピックの最後のメッセージを保存して、<br>
新しいクライアントが購読を開始した時に、最後に発行されたメッセージを自動的に送信するかどうかを制御するフラグである。<br>
<br>
publishメソッドでは、以下に示す値を指定することができる。<br>
* RETAIN (true/1):
*: メッセージをMQTTブローカーに保持するように指示する。
*: 新しいサブスクライバーが接続した時に、最後のRetainメッセージを受信する。
*: 頻繁に更新されないステータス情報 (デバイスの設定値等) の共有に使用する。
*: <br>
* NO_RETAIN (false/0)
*: メッセージは保持されない。
*: 購読時に過去のメッセージは送信されない。
*: リアルタイムデータやイベント通知等の一時的な情報に適している。
<br>
<syntaxhighlight lang="c++">
// 使用例:
QMqttClient client;
// 保持フラグをtrueに設定してメッセージを発行 (QoS 0, Retain true)
client.publish("sensor/temperature", "25.5", 0, true);
// 保持フラグをfalseに設定してメッセージを発行 (QoS 0, Retain false)
client.publish("sensor/temperature", "25.5", 0, false);
</syntaxhighlight>
<br>
MQTTブローカーに保持されているメッセージを削除する場合は、同じトピックに空のペイロードを保持フラグtrueで送信する。<br>
<syntaxhighlight lang="c++">
QMqttClient client;
client.publish("sensor/temperature", "", 0, true);
</syntaxhighlight>
<br>
==== 送信 (パブリッシュ側) ====
<syntaxhighlight lang="c++">
QMqttClient client;
// メッセージのパブリッシュ
qint32 msgId = client->publish(topic,    // トピック名
                                message,  // メッセージ内容
                                qos,      // QoSレベル
                                retain);  // 保持メッセージフラグ
// 戻り値が-1の場合はパブリッシュ失敗
if (msgId == -1) {
    // エラー処理
    // ...略
}
</syntaxhighlight>
<br>
==== 受信 (サブスクライブ側) ====
<syntaxhighlight lang="c++">
QMqttClient client;
// トピックのサブスクライブ
// QoS 0 : (最大1回配信)
// QoS 1 : (最低1回配信)
// QoS 2 : (正確に1回配信)
auto subscription = client->subscribe(topic, qos);
// サブスクリプション成功の確認
if (subscription) {
    QMqttSubscription *sub = subscription;
    // サブスクリプションの状態変更を監視
    connect(sub, &QMqttSubscription::stateChanged, this, [](QMqttSubscription::SubscriptionState state) {
      // 状態の種類:
      // -> QMqttSubscription::Unsubscribed
      // -> QMqttSubscription::SubscriptionPending
      // -> QMqttSubscription::Subscribed
      // -> QMqttSubscription::UnsubscriptionPending
    });
}
</syntaxhighlight>
<br>
==== シグナル / スロット接続 ====
* 接続状態の変更を監視する場合
<br>
<syntaxhighlight lang="c++">
connect(client, &QMqttClient::stateChanged, this, [](QMqttClient::ClientState state) {
    // 状態変更時の処理
    // ...略
});
</syntaxhighlight>
<br>
* メッセージを受信する場合
<br>
<syntaxhighlight lang="c++">
connect(client, &QMqttClient::messageReceived, this, [](const QByteArray &message, const QMqttTopicName &topic) {
    // メッセージの受信
    QString msg = QString::fromUtf8(message);
    QString topicName = topic.name();
    // その他のメッセージ処理
    // ...略
});
</syntaxhighlight>
<br>
<syntaxhighlight lang="c++">
// エラー監視
connect(client, &QMqttClient::errorChanged, this, [](QMqttClient::ClientError error) {
    // エラーの種類:
    // -> QMqttClient::NoError
    // -> QMqttClient::InvalidProtocolVersion
    // -> QMqttClient::IdRejected
    // -> QMqttClient::ServerUnavailable
    // -> QMqttClient::BadUsernameOrPassword
    // -> QMqttClient::NotAuthorized
    // -> QMqttClient::TransportInvalid
    // -> QMqttClient::ProtocolViolation
    // -> QMqttClient::UnknownError
    // -> QMqttClient::Mqtt5SpecificError
    // エラー処理を記述
    // ...略
});
</syntaxhighlight>
<br><br>
<br><br>


38行目: 238行目:
また、シリアルモニタを活用して、接続状態やメッセージの送受信を確認する。<br>
また、シリアルモニタを活用して、接続状態やメッセージの送受信を確認する。<br>
<br>
<br>
==== CMake / Qtプロジェクトファイル ====
* CMakeLists.txtファイルを使用する場合
<syntaxhighlight lang="cmake">
# パッケージの検索
find_package(QT NAMES Qt6 REQUIRED COMPONENTS Core Network Mqtt)
find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Network Mqtt)
# ライブラリのリンク
target_link_libraries(<プロジェクト名> PRIVATE
    Qt6::Core
    Qt6::Network
    Qt6::Mqtt
) </syntaxhighlight>
<br>
* Qtプロジェクトファイルを使用する場合
<syntaxhighlight lang="make">
QT += core network mqtt
</syntaxhighlight>
<br>
==== 使用例 ====
==== 使用例 ====
以下の例では、MQTT通信でトピックの送受信を行っている。<br>
以下の例では、MQTT通信でトピックの送受信を行っている。<br>
59行目: 279行目:
  private:
  private:
     QMqttClient *m_client;
     QMqttClient *m_client;
     QString m_host;
     QString     m_host;
     quint16 m_port;
     quint16     m_port;
   
   
  public:
  public:
88行目: 308行目:
       m_client->setHostname(host);
       m_client->setHostname(host);
       m_client->setPort(port);
       m_client->setPort(port);
      // ユーザ名とパスワードが指定されている場合は設定
      if (!username.isEmpty()) {
          m_client->setUsername(username);
      }
      if (!password.isEmpty()) {
          m_client->setPassword(password);
      }
   
   
       // 非同期で接続を開始
       // 非同期で接続を開始
104行目: 333行目:
       auto publish = m_client->publish(topic, message, 1);
       auto publish = m_client->publish(topic, message, 1);
       if (!publish) {
       if (!publish) {
           qWarning() << "メッセージの発行に失敗しました: " << topic;
           qWarning() << "メッセージの発行に失敗しました : " << topic;
           return;
           return;
       }
       }
116行目: 345行目:
       }
       }
     }
     }
signals:
    // 接続確立時のシグナルを追加
    void connectionEstablished();
   
   
  private slots:
  private slots:
124行目: 357行目:
           case QMqttClient::Connected:
           case QMqttClient::Connected:
             qInfo() << "MQTTブローカーに接続しました";
             qInfo() << "MQTTブローカーに接続しました";
            emit connectionEstablished();  // 接続確立時にシグナルを発行
             break;
             break;
           case QMqttClient::Disconnected:
           case QMqttClient::Disconnected:
173行目: 407行目:
       }
       }
   
   
       qCritical() << "MQTTエラー:" << errorMsg;
       qCritical() << "MQTTエラー :" << errorMsg;
   
   
       // エラー発生時は再接続を試みる
       // エラー発生時は再接続を試みる
       if (m_client->state() != QMqttClient::Connected) {
       if (m_client->state() != QMqttClient::Connected) {
           QTimer::singleShot(5000, [this]() {
           QTimer::singleShot(5000, this, [this]() {
             qInfo() << "再接続を試みます...";
             qInfo() << "再接続を試みます...";
             this->connectToHost(m_host, m_port);
             this->connectToHost(m_host, m_port, m_client->username(), m_client->password());
           });
           });
       }
       }
196行目: 430行目:
   
   
     MqttPublisher publisher;
     MqttPublisher publisher;
    publisher.connectToHost("<IPアドレスまたはホスト名  例: localhost>", <MQTT通信するポート番号  例: 1883>);
   
   
     // 定期的にメッセージを送信
     // タイマの作成
     QTimer timer;
     QTimer *pTimer = new QTimer(&a);
     QObject::connect(&timer, &QTimer::timeout, [&publisher]() {
 
    // タイマーのコールバックを設定
     QObject::connect(pTimer, &QTimer::timeout, pTimer, [&publisher]() {
       publisher.publishMessage("qt/topic", "Hello, MQTT!");
       publisher.publishMessage("qt/topic", "Hello, MQTT!");
     });
     });
 
     timer.start(1000);  // 1秒おきに送信
     // Publisher側で接続状態の変化を監視し、接続成功時にタイマーを開始
    QObject::connect(&publisher, &Publisher::connectionEstablished, pTimer, [pTimer]() {
      pTimer->start(3000);  // 3秒おきに送信
    });
 
    // MQTTブローカーに接続
    publisher.connectToHost("<IPアドレスまたはホスト名  例: localhost>",
                            <MQTT通信するポート番号  例: 1883>,
                            "<MQTTユーザ名 (匿名ユーザを許可している場合は空でも可)>",
                            "<MQTTユーザのパスワード (匿名ユーザを許可している場合は空でも可>");
      
      
     return a.exec();
     return a.exec();
245行目: 489行目:
     {
     {
       // 全ての購読を解除
       // 全ての購読を解除
       for (auto subscription : m_subscriptions) {
       for (auto it = m_subscriptions.constBegin(); it != m_subscriptions.constEnd(); it++) {
           subscription->unsubscribe();
           it.value()->unsubscribe();
           delete subscription;
           delete it.value();
       }
       }
   
   
258行目: 502行目:
   
   
     // MQTT接続を開始する
     // MQTT接続を開始する
     void connectToHost(const QString &host, quint16 port)
     void connectToHost(const QString &host, quint16 port, const QString &username = QString(), const QString &password = QString())
     {
     {
       m_host = host;
       m_host = host;
266行目: 510行目:
       m_client->setHostname(host);
       m_client->setHostname(host);
       m_client->setPort(port);
       m_client->setPort(port);
      // ユーザ名とパスワードが指定されている場合は設定
      if (!username.isEmpty()) {
          m_client->setUsername(username);
      }
      if (!password.isEmpty()) {
          m_client->setPassword(password);
      }
   
   
       // 非同期で接続を開始
       // 非同期で接続を開始
288行目: 541行目:
       auto subscription = m_client->subscribe(topic, 1);
       auto subscription = m_client->subscribe(topic, 1);
       if (!subscription) {
       if (!subscription) {
           qWarning() << "トピックの購読に失敗しました:" << topic;
           qWarning() << "トピックの購読に失敗しました : " << topic;
           return;
           return;
       }
       }
297行目: 550行目:
       // 購読リストに追加
       // 購読リストに追加
       m_subscriptions.insert(topic, subscription);
       m_subscriptions.insert(topic, subscription);
       qInfo() << "トピックを購読開始しました:" << topic;
       qInfo() << "トピックを購読開始しました : " << topic;
     }
     }
   
   
309行目: 562行目:
   
   
  signals:
  signals:
     // メッセージ受信時のシグナル
     void connected();
     void messageReceived(const QString &topic, const QByteArray &message);
     void messageReceived(const QString &topic, const QByteArray &message);
   
   
319行目: 572行目:
           case QMqttClient::Connected:
           case QMqttClient::Connected:
             qInfo() << "MQTTブローカーに接続しました";
             qInfo() << "MQTTブローカーに接続しました";
            emit connected();  // 接続成功時にシグナルを発行
             break;
             break;
           case QMqttClient::Disconnected:
           case QMqttClient::Disconnected:
             qInfo() << "MQTTブローカーから切断されました";
             qInfo() << "MQTTブローカーから切断されました";
            // 切断時に再接続を試みる
            QTimer::singleShot(5000, this, [this]() {
                qInfo() << "再接続を試みます...";
                this->connectToHost(m_host, m_port, m_client->username(), m_client->password());
            });
             break;
             break;
           case QMqttClient::Connecting:
           case QMqttClient::Connecting:
372行目: 633行目:
       // エラー発生時は再接続を試みる
       // エラー発生時は再接続を試みる
       if (m_client->state() != QMqttClient::Connected) {
       if (m_client->state() != QMqttClient::Connected) {
           QTimer::singleShot(5000, [this]() {
           QTimer::singleShot(5000, this, [this]() {
             qInfo() << "再接続を試みます...";
             qInfo() << "再接続を試みます...";
             this->connectToHost(m_host, m_port);
             this->connectToHost(m_host, m_port, m_client->username(), m_client->password());
           });
           });
       }
       }
382行目: 643行目:
     void handleMessage(const QByteArray &message, const QMqttTopicName &topic)
     void handleMessage(const QByteArray &message, const QMqttTopicName &topic)
     {
     {
       // メッセージを受信したことをログに記録
       // シグナルを発行してアプリケーションに通知
      qInfo() << "メッセージを受信しました - トピック: " << topic.name();
      qInfo() << "メッセージ: " << message;
      // シグナルを発行して、アプリケーションの他の部分に通知
       emit messageReceived(topic.name(), message);
       emit messageReceived(topic.name(), message);
     }
     }
402行目: 659行目:
   
   
     MqttSubscriber subscriber;
     MqttSubscriber subscriber;
    subscriber.connectToHost("localhost", 1883);
   
   
     // メッセージ受信時の処理
     // メッセージ受信時の処理
     QObject::connect(&subscriber, &MqttSubscriber::messageReceived, [](const QString &topic, const QByteArray &message) {
     QObject::connect(&subscriber, &Subscriber::messageReceived, [](const QString &topic, const QByteArray &message) {
       qDebug() << "受信 - トピック: " << topic;
       qDebug() << "受信 - トピック : " << topic;
       qDebug() << "メッセージ: "  << message;
       qDebug() << "メッセージ : "  << message;
     });
     });
   
   
     subscriber.subscribe("qt/topic");
     // 接続成功時の処理を追加
    QObject::connect(&subscriber, &Subscriber::connected, [&subscriber]() {
      subscriber.subscribe("qt/topic");
    });
    subscriber.connectToHost("localhost", 1883, "<MQTTユーザ名>", "<MQTTユーザのパスワード>");
    return a.exec();
}
</syntaxhighlight>
<br>
===== 送受信 (MQTTパブリッシャー / MQTTサブスクライバ) =====
以下の例では、トピックの送受信を行っている。<br>
<br>
* クラス構造
** MqttWorker
**: MQTT通信の実際の処理を担当
** MqttManager
**: アプリケーションとワーカー間のインターフェース
*: <br>
* スレッド管理
*: MQTTの処理を専用スレッドで実行
*: シグナル / スロットによる安全な通信
*: <br>
* エラーハンドリング
*: 全てのエラーはerrorOccurredシグナルで通知
*: 接続問題は自動的に処理
*: 包括的なエラー状態の検出と通知
*: デバッグ情報の出力
*: <br>
* 非同期処理
*: 全ての操作が非ブロッキング
*: メッセージキューイングによる信頼性の確保
<br>
<syntaxhighlight lang="c++">
// Mqttworker.h
#include <QObject>
#include <QtMqtt/QMqttClient>
#include <QThread>
#include <QQueue>
#include <QMutex>
/**
  * @brief MQTTの通信処理を担当するワーカークラス
  *
  * このクラスは別スレッドで動作して、MQTT通信に関する以下に示す機能を提供する
  * - MQTTブローカーへの接続 / 切断
  * - メッセージの発行 (パブリッシュ)
  * - トピックの購読（サブスクライブ）
  * - メッセージの受信と通知
  *
  * スレッドセーフな設計となっており、メッセージキューイング機能も備えている
  */
class MqttWorker : public QObject
{
    Q_OBJECT
private:
    QMqttClient *m_client;                                  // MQTTクライアントインスタンス
    QQueue<QPair<QString, QByteArray>> m_messageQueue;      // 未送信メッセージのキュー
    QMutex m_mutex;                                        // スレッド同期用ミューテックス
    bool m_isConnected;                                    // 現在の接続状態
public:
    /**
    * @brief コンストラクタ
    * @param parent 親オブジェクト（デフォルトはnullptr）
    */
    explicit MqttWorker(QObject *parent = nullptr) : QObject(parent), m_client(new QMqttClient(this)), m_isConnected(false)
    {
      // MQTTクライアントのシグナルとワーカーのスロットを接続
      connect(m_client, &QMqttClient::messageReceived, this, &MqttWorker::handleMessage);
      connect(m_client, &QMqttClient::stateChanged, this, &MqttWorker::handleStateChange);
      connect(m_client, &QMqttClient::errorChanged, this, &MqttWorker::handleError);
    }
    /**
    * @brief デストラクタ
    * 接続中の場合は切断処理を行います
    */
    ~MqttWorker()
    {
      if (m_client->state() == QMqttClient::Connected) {
          m_client->disconnectFromHost();
      }
    }
public slots:
    /**
    * @brief MQTTブローカーへの接続を開始
    * @param host ブローカーのホスト名またはIPアドレス
    * @param port ブローカーのポート番号
    *
    * このメソッドは非同期で実行され、接続状態の変更はconnectionStateChangedシグナルで通知される
    */
    void connectToHost(const QString &host, quint16 port)
    {
      qDebug() << "ワーカースレッド開始: " << QThread::currentThread();
      m_client->setHostname(host);
      m_client->setPort(port);
      m_client->connectToHost();
    }
    /**
    * @brief 指定されたトピックにメッセージを発行
    * @param topic 発行先のトピック
    * @param payload 送信するメッセージ内容
    *
    * 未接続時はメッセージをキューに保存し、接続時に自動的に送信する
    * QoS 1を使用して、メッセージの到達を保証する
    */
    void publishMessage(const QString &topic, const QByteArray &payload)
    {
      QMutexLocker locker(&m_mutex);
      if (!m_isConnected) {
          // 未接続時はメッセージをキューに保存
          m_messageQueue.enqueue(qMakePair(topic, payload));
          qDebug() << "メッセージをキューに保存: " << topic;
          return;
      }
      auto publish = m_client->publish(topic, payload, 1);
      if (!publish) {
          emit errorOccurred("メッセージの発行に失敗しました: " + topic);
          qWarning() << "メッセージ発行失敗: " << topic;
      }
      else {
          qDebug() << "メッセージを発行:" << topic;
      }
    }
    /**
    * @brief 指定されたトピックを購読
    * @param topic 購読するトピック名
    *
    * 接続済みの場合のみ購読を開始する
    * QoS 1を使用して、メッセージの到達を保証する
    */
    void subscribe(const QString &topic)
    {
      if (!m_isConnected) {
          emit errorOccurred("購読できません - 接続されていません");
          qWarning() << "購読失敗 - 未接続:" << topic;
          return;
      }
      auto subscription = m_client->subscribe(topic, 1);
      if (!subscription) {
          emit errorOccurred("トピックの購読に失敗しました: " + topic);
          qWarning() << "購読失敗:" << topic;
      }
      else {
          qDebug() << "トピックを購読開始:" << topic;
      }
    }
signals:
    /**
    * @brief メッセージ受信時に発行されるシグナル
    * @param topic 受信したメッセージのトピック
    * @param message 受信したメッセージの内容
    */
    void messageReceived(const QString &topic, const QByteArray &message);
    /**
    * @brief 接続状態が変化した時に発行されるシグナル
    * @param connected 接続状態 (true: 接続済み, false: 未接続)
    */
    void connectionStateChanged(bool connected);
    /**
    * @brief エラー発生時に発行されるシグナル
    * @param error エラーメッセージ
    */
    void errorOccurred(const QString &error);
private slots:
    /**
    * @brief メッセージ受信時の処理
    * @param message 受信したメッセージ
    * @param topic 受信したトピック
    */
    void handleMessage(const QByteArray &message, const QMqttTopicName &topic) {
        emit messageReceived(topic.name(), message);
        qDebug() << "メッセージを受信:" << topic.name() << message;
    }
 
    /**
    * @brief 接続状態変更時の処理
    *
    * 接続完了時にキューに保存されたメッセージの送信を試みます
    */
    void handleStateChange()
    {
      m_isConnected = (m_client->state() == QMqttClient::Connected);
      emit connectionStateChanged(m_isConnected);
      if (m_isConnected) {
          qDebug() << "MQTT接続完了";
          // 接続時にキューのメッセージを処理
          QMutexLocker locker(&m_mutex);
          while (!m_messageQueue.isEmpty()) {
            auto message = m_messageQueue.dequeue();
            publishMessage(message.first, message.second);
          }
      }
      else {
          qDebug() << "MQTT切断";
      }
    }
    /**
    * @brief エラー発生時の処理
    * @param error 発生したエラーの種類
    *
    * エラーの種類に応じて適切なメッセージを生成し通知します
    */
    void handleError(QMqttClient::ClientError error)
    {
      QString errorMessage;
      switch (error) {
          case QMqttClient::NoError:
            return;
          case QMqttClient::InvalidProtocolVersion:
            errorMessage = "無効なプロトコルバージョン";
            break;
          case QMqttClient::IdRejected:
            errorMessage = "クライアントID拒否";
            break;
          case QMqttClient::ServerUnavailable:
            errorMessage = "サーバー利用不可";
            break;
          case QMqttClient::BadUsernameOrPassword:
            errorMessage = "認証エラー";
            break;
          case QMqttClient::NotAuthorized:
            errorMessage = "認可エラー";
            break;
          case QMqttClient::TransportInvalid:
            errorMessage = "トランスポートエラー";
            break;
          case QMqttClient::ProtocolViolation:
            errorMessage = "プロトコル違反";
            break;
          case QMqttClient::UnknownError:
            errorMessage = "不明なエラー";
            break;
          default:
            errorMessage = "予期せぬエラー";
            break;
      }
      emit errorOccurred(errorMessage);
      qCritical() << "MQTTエラー:" << errorMessage;
    }
};
</syntaxhighlight>
<br>
<syntaxhighlight lang="c++">
// Mqttmanager.hファイル
#include <QObject>
#include <QThread>
#include "Mqttworker.h"
/**
  * @brief MQTT通信を管理するマネージャークラス
  *
  * このクラスは以下に示す機能を提供する
  * - MQTTワーカーの生成と管理
  * - ワーカースレッドの制御
  * - MQTT通信操作のインターフェース提供
  *
  * アプリケーションはこのクラスを通じてMQTT通信を利用する
  * 全ての操作は非同期で実行されて、メインスレッドをブロックしない
  */
class MqttManager : public QObject
{
    Q_OBJECT
private:
    QThread m_workerThread;  // ワーカー用スレッド
    MqttWorker *m_worker;    // MQTTワーカーインスタンス
public:
    /**
    * @brief コンストラクタ
    * @param parent 親オブジェクト（デフォルトはnullptr）
    *
    * ワーカーオブジェクトを生成して、専用スレッドで実行を開始する
    * 必要なシグナル / スロット接続も確立する
    */
    explicit MqttManager(QObject *parent = nullptr) : QObject(parent), m_worker(new MqttWorker)
    {
      // ワーカーを別スレッドに移動
      m_worker->moveToThread(&m_workerThread);
      // マネージャーからワーカーへのシグナル接続
      connect(this, &MqttManager::connectRequested, m_worker, &MqttWorker::connectToHost);
      connect(this, &MqttManager::publishRequested, m_worker, &MqttWorker::publishMessage);
      connect(this, &MqttManager::subscribeRequested, m_worker, &MqttWorker::subscribe);
      // ワーカーからマネージャーへのシグナル接続
      connect(m_worker, &MqttWorker::messageReceived, this, &MqttManager::messageReceived);
      connect(m_worker, &MqttWorker::connectionStateChanged, this, &MqttManager::connectionStateChanged);
      connect(m_worker, &MqttWorker::errorOccurred, this, &MqttManager::errorOccurred);
      // スレッド終了時のクリーンアップ設定
      connect(&m_workerThread, &QThread::finished, m_worker, &MqttWorker::deleteLater);
      // ワーカースレッドを開始
      m_workerThread.start();
      qDebug() << "MQTTマネージャー初期化完了";
    }
    /**
    * @brief デストラクタ
    *
    * ワーカースレッドを適切に終了して、リソースを解放する
    */
    ~MqttManager()
    {
      m_workerThread.quit();
      m_workerThread.wait();
      qDebug() << "MQTTマネージャー終了";
    }
    /**
    * @brief MQTTブローカーへの接続を要求
    * @param host ブローカーのホスト名またはIPアドレス
    * @param port ブローカーのポート番号
    */
    void connect(const QString &host, quint16 port)
    {
        emit connectRequested(host, port);
        qDebug() << "接続要求:" << host << port;
    }
    /**
    * @brief メッセージの発行を要求
    * @param topic 発行先のトピック
    * @param payload 送信するメッセージ内容
    */
    void publish(const QString &topic, const QByteArray &payload)
    {
      emit publishRequested(topic, payload);
      qDebug() << "発行要求:" << topic;
    }
    /**
    * @brief トピックの購読を要求
    * @param topic 購読するトピック名
    */
    void subscribe(const QString &topic)
    {
        emit subscribeRequested(topic);
        qDebug() << "購読要求:" << topic;
    }
signals:
    // ワーカーへの要求シグナル
    void connectRequested(const QString &host, quint16 port);              // 接続要求シグナル
    void publishRequested(const QString &topic, const QByteArray &payload); // 発行要求シグナル
    void subscribeRequested(const QString &topic);                          // 購読要求シグナル
    // アプリケーションへの通知シグナル
    void messageReceived(const QString &topic, const QByteArray &message);  // メッセージ受信通知
    void connectionStateChanged(bool connected);                            // 接続状態変更通知
    void errorOccurred(const QString &error);                              // エラー発生通知
};
</syntaxhighlight>
<br>
<syntaxhighlight lang="c++">
// 使用例 : main.cppファイル
#include <QCoreApplication>
#include "Mqttmanager.h"
int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);
    // MQTTマネージャーのインスタンスを生成
    MqttManager manager;
    // 接続状態変更時の処理
    QObject::connect(&manager, &MqttManager::connectionStateChanged, [](bool connected) {
      if (connected) {
          qDebug() << "MQTT接続完了";
      }
      else {
          qDebug() << "MQTT切断";
      }
    });
    // メッセージ受信時の処理
    QObject::connect(&manager, &MqttManager::messageReceived, [](const QString &topic, const QByteArray &message) {
      qDebug() << "メッセージ受信 - トピック: " << topic;
      qDebug() << "内容: " << message;
    });
    // エラー発生時の処理
    QObject::connect(&manager, &MqttManager::errorOccurred, [](const QString &error) {
      qCritical() << "MQTTエラー: " << error;
    });
    // MQTTブローカーへの接続
    manager.connect("<IPアドレスまたはホスト名  例: localhost>", <ポート番号  例: 1883>);
    // トピックの購読
    manager.subscribe("qt/topic");
    // メッセージの発行
    QTimer::singleShot(1000, [&manager]() {
      manager.publish("qt/topic", "Hello、MQTT!");
    });
   
   
     return a.exec();
     return a.exec();
「http://mochiuwiki.e2.valueserver.jp/index.php?title=Qtの基礎_-_MQTT」から取得