MochiuWiki : SUSE, EC, PCB

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

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

← 古い編集
ビジュアルウィキテキスト
 
(同じ利用者による、間の4版が非表示)
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>


42行目: 242行目:
  <syntaxhighlight lang="cmake">
  <syntaxhighlight lang="cmake">
  # パッケージの検索
  # パッケージの検索
  find_package(Qt6 COMPONENTS Core Network Mqtt REQUIRED)
  find_package(QT NAMES Qt6 REQUIRED COMPONENTS Core Network Mqtt)
  find_package(PkgConfig REQUIRED)
  find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Network Mqtt)
# # Pkg-configを使用してMQTTライブラリを検索
pkg_check_modules(MQTT REQUIRED IMPORTED_TARGET
    libmosquitto
    libmosquittopp
)
   
   
  # ライブラリのリンク
  # ライブラリのリンク
56行目: 250行目:
     Qt6::Network
     Qt6::Network
     Qt6::Mqtt
     Qt6::Mqtt
    PkgConfig::MQTT
  ) </syntaxhighlight>
  )
# インクルードするディレクトリの設定
target_include_directories(<プロジェクト名> PRIVATE
    # ...略
    ${MQTT_INCLUDE_DIRS}
)
</syntaxhighlight>
<br>
<br>
* Qtプロジェクトファイルを使用する場合
* Qtプロジェクトファイルを使用する場合
  <syntaxhighlight lang="make">
  <syntaxhighlight lang="make">
  QT += core network mqtt
  QT += core network mqtt
# Pkg-configを使用してMQTTライブラリを検索
CONFIG += link_pkgconfig
PKGCONFIG += libmosquitto libmosquittopp
  </syntaxhighlight>
  </syntaxhighlight>
<br>
<br>
97行目: 279行目:
  private:
  private:
     QMqttClient *m_client;
     QMqttClient *m_client;
     QString m_host;
     QString     m_host;
     quint16 m_port;
     quint16     m_port;
   
   
  public:
  public:
126行目: 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);
      }
   
   
       // 非同期で接続を開始
       // 非同期で接続を開始
142行目: 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;
       }
       }
154行目: 345行目:
       }
       }
     }
     }
signals:
    // 接続確立時のシグナルを追加
    void connectionEstablished();
   
   
  private slots:
  private slots:
162行目: 357行目:
           case QMqttClient::Connected:
           case QMqttClient::Connected:
             qInfo() << "MQTTブローカーに接続しました";
             qInfo() << "MQTTブローカーに接続しました";
            emit connectionEstablished();  // 接続確立時にシグナルを発行
             break;
             break;
           case QMqttClient::Disconnected:
           case QMqttClient::Disconnected:
211行目: 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());
           });
           });
       }
       }
234行目: 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();
283行目: 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();
       }
       }
   
   
296行目: 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;
304行目: 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);
      }
   
   
       // 非同期で接続を開始
       // 非同期で接続を開始
326行目: 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;
       }
       }
335行目: 550行目:
       // 購読リストに追加
       // 購読リストに追加
       m_subscriptions.insert(topic, subscription);
       m_subscriptions.insert(topic, subscription);
       qInfo() << "トピックを購読開始しました:" << topic;
       qInfo() << "トピックを購読開始しました : " << topic;
     }
     }
   
   
347行目: 562行目:
   
   
  signals:
  signals:
     // メッセージ受信時のシグナル
     void connected();
     void messageReceived(const QString &topic, const QByteArray &message);
     void messageReceived(const QString &topic, const QByteArray &message);
   
   
357行目: 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:
410行目: 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());
           });
           });
       }
       }
420行目: 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);
     }
     }
440行目: 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;
    });
    // 接続成功時の処理を追加
    QObject::connect(&subscriber, &Subscriber::connected, [&subscriber]() {
      subscriber.subscribe("qt/topic");
     });
     });
   
   
     subscriber.subscribe("qt/topic");
     subscriber.connectToHost("localhost", 1883, "<MQTTユーザ名>", "<MQTTユーザのパスワード>");
   
   
     return a.exec();
     return a.exec();
「http://mochiuwiki.e2.valueserver.jp/index.php?title=Qtの基礎_-_MQTT」から取得