他の MQTT サービスとのブリッジ

MQTT ブローカーのデータ統合は、EMQX に別の EMQX クラスターや他の MQTT サービスに接続してメッセージブリッジを実現する機能を提供し、ネットワークやサービスを跨いだデータの相互作用と通信を可能にします。このページでは、EMQX における MQTT メッセージブリッジの動作原理を紹介し、メッセージブリッジの作成と検証に関する実践的なガイドを提供します。

動作原理

ブリッジング中、EMQX はクライアントとしてターゲットサービスと MQTT 接続を確立し、パブリッシュ・サブスクライブモデルを通じて双方向のメッセージ送受信を実現します。

• 送信メッセージ（Sink）：ローカルのトピックからリモート MQTT サービスの指定トピックにメッセージをパブリッシュします。
• 受信メッセージ（Source）：リモート MQTT サービスのトピックをサブスクライブし、そのメッセージをローカルの EMQX に転送します。

EMQX は同一接続上で複数のブリッジルールを設定可能で、それぞれ異なるトピックマッピングやメッセージ変換ルールを持ち、メッセージルーティングに類似した機能を実現します。ブリッジング中は、ルールエンジンを通じてメッセージのフィルタリング、拡充、変換処理も行えます。

以下の図は、EMQX と他の MQTT サービス間のデータ統合の典型的なアーキテクチャを示しています。

特徴と利点

MQTT ブローカーのデータ統合は以下の特徴と利点を持ちます。

• 高い互換性：標準 MQTT プロトコルを使用しているため、AWS IoT Core、Azure IoT Hub を含む多様な IoT プラットフォームとブリッジ可能であり、オープンソースや他の業界標準 MQTT ブローカーや IoT プラットフォームもサポートします。これにより、多様なデバイスやプラットフォームとのシームレスな統合と通信が可能です。
• 双方向データフロー：EMQX からリモート MQTT サービスへのメッセージパブリッシュと、リモート MQTT サービスからのメッセージサブスクライブおよびローカルパブリッシュの双方向通信をサポートします。この双方向通信機能により、異なるシステム間のデータ転送が柔軟かつ制御可能になります。
• 柔軟なトピックマッピング：MQTT のパブリッシュ・サブスクライブモデルに基づき、トピックにプレフィックスを付加したり、クライアントのコンテキスト情報（クライアントID、ユーザー名など）を用いて動的にトピックを構築したりする柔軟なトピックマッピングを実装しています。これにより、ニーズに応じたメッセージのカスタマイズ処理やルーティングが可能です。
• 高性能：コネクションプーリングや共有サブスクリプションなどのパフォーマンス最適化オプションを提供し、個別のブリッジクライアントの負荷を軽減、ブリッジングのレイテンシ低減とメッセージスループットの向上を実現します。これらの最適化により、システム全体のパフォーマンスとスケーラビリティが向上します。
• ペイロード変換：SQL ルールを定義してメッセージペイロードの処理が可能です。メッセージ送信時にデータ抽出、フィルタリング、拡充、変換などの操作をペイロードに対して行えます。例えば、リアルタイムのメトリクスをペイロードから抽出し、変換・加工してからリモート MQTT ブローカーに届けることができます。
• メトリクス監視：各 Sink/Source ごとにランタイムのメトリクス監視を提供し、総メッセージ数、成功/失敗数、現在のレートなどを確認可能です。これにより、Sink/Source のパフォーマンスや健全性をリアルタイムに監視・評価できます。

MQTT 接続情報の準備

前提条件

以下を事前に理解しておいてください。

• ルールエンジン
• データ統合

MQTT ブローカーのデータ統合を作成する前に、リモート MQTT サービスの接続情報を取得する必要があります。ここでは EMQX のオンライン MQTT サーバーを例に説明します。

• MQTT サービスアドレス：ターゲット MQTT サービスのアドレスとポート。例では broker.emqx.io:1883。
  • 対応フォーマットは host:port、[IPv6]:port、mqtt://host:port、mqtt://[IPv6]:port、mqtts://host:port、mqtts://[IPv6]:port です。ポート省略時はデフォルト MQTT ポート 1883 を使用します。その他の URI スキームは非対応です。
  • TLS はコネクター設定で別途構成します。アドレススキームと TLS 設定の関係は以下の通りです。
    • アドレスが明示的に mqtts:// を使う場合は TLS を有効にする必要があります。そうしないと 接続テスト が失敗し、コネクターは起動しません。
    • アドレスが mqtt:// で TLS が有効の場合は TLS 設定が優先され、EMQX は TLS 接続し警告をログに記録します。
    • スキームなしのアドレスの場合は TLS 設定に応じて TLS を使用するか決まります。
• ユーザー名：接続に必要なユーザー名。認証不要の場合は空欄可。
• パスワード：接続に必要なパスワード。認証不要の場合は空欄可。
• プロトコルタイプ：ターゲットサービスが TLS を有効にしているか、TCP/TLS 上の MQTT を使っているかを確認してください。EMQX の MQTT ブリッジは現在、MQTT over WebSocket や MQTT over QUIC などのプロトコルはサポートしていません。
• プロトコルバージョン：ターゲット MQTT サービスのプロトコルバージョン。EMQX は MQTT 3.1、3.1.1、5.0 をサポートしています。

データ統合は EMQX や他の標準 MQTT サーバーとの互換性とサポートが良好です。その他のタイプの MQTT サービスに接続する場合は、該当サービスのドキュメントを参照して接続情報を取得してください。一般に多くの IoT プラットフォームは標準 MQTT アクセス方法を提供しており、それに従ってデバイス情報を上記の MQTT 接続情報に変換できます。

注意

EMQX がクラスター運用中、またはコネクションプールを有効にしている場合、同一クライアントIDを使って複数ノードが同じ MQTT サービスに接続するとデバイス競合が発生しやすいため、MQTT メッセージブリッジでは固定クライアントIDの設定はサポートしていません。

コネクターの作成

このセクションでは、リモート MQTT サーバーとの接続を設定する方法を案内します。

1. ダッシュボードの Integration -> Connector ページに移動します。

2. ページ右上の Create ボタンをクリックします。

3. コネクタータイプ一覧から MQTT Broker を選択し、Next をクリックします。

4. コネクターの 名前 を入力します。英数字の組み合わせで、例として my_mqtt_bridge などを指定します。

5. 接続情報を設定します：

   • MQTT Broker：MQTT コネクターは TCP/TLS 上の MQTT のみサポートします。MQTT サービスアドレスを入力してください。例：broker.emqx.io:1883、[::1]:1883、mqtt://broker.emqx.io:1883。
   • TLS を有効にする：TLS 対応の MQTT リスナーに接続する場合は有効にします。アドレスが明示的に mqtts:// の場合は必ず有効にしてください。スキームなしのアドレスの場合、この設定が TLS 利用の有無を決定します。
   • ClientID プレフィックス：空欄可。実際の運用ではクライアントIDプレフィックスを指定するとクライアント管理が容易になります。EMQX はプレフィックスとコネクションプールサイズに基づきクライアントIDを自動生成します。詳細は コネクションプールとクライアントID生成ルール を参照してください。
   • ユーザー名 と パスワード：MQTT ブローカーが認証を要求する場合は、クライアントIDに紐づくユーザー名とパスワードを入力します。認証不要（パブリックブローカーなど）の場合は空欄可。
   • キープアライブ：希望するキープアライブ間隔を指定します。
   • MQTT バージョン：ブローカー接続に適したバージョンを選択します。
   • 静的 ClientId エントリー：Azure IoT Hub など特定サービスに接続する際、安定した接続のために静的クライアントIDを設定できます。設定方法は 静的クライアントIDの設定 を参照してください。

   TIP

   静的クライアントIDエントリーを定義した場合、静的クライアントIDが割り当てられた EMQX ノードのみが MQTT 接続を開始します。

その他の設定はデフォルトのままにし、Create ボタンをクリックしてコネクター作成を完了します。作成したコネクターは Sink と Source の両方で利用可能です。次に、このコネクターを使ってデータブリッジルールを作成できます。

コネクションプールとクライアントID生成ルール

EMQX は複数のクライアントが同時にブリッジ対象の MQTT サービスに接続できるようにします。コネクター作成時に MQTT クライアントのコネクションプールを設定し、そのサイズを指定できます。コネクションプールはサーバーリソースを最大限活用し、メッセージスループットと同時接続性能を向上させ、高負荷・高並列シナリオに対応します。

MQTT プロトコルは MQTT サーバーに接続するクライアントにユニークなクライアントIDを要求します。EMQX はクラスター展開可能なため、MQTT ブリッジの各クライアントにユニークなクライアントIDを割り当てます。クライアントIDは以下のパターンで自動生成されます。

[Client ID Prefix]:{Connector Name}{8桁のランダム文字列}:{プール内の接続シーケンス番号}

例として、クライアントIDプレフィックスが myprefix、コネクター名が foo の場合、実際のクライアントIDは以下のようになります。

myprefix:foo2bd61c44:1

バージョン 5.4.1 以降、EMQX は MQTT クライアントIDの長さを最大 23 バイトに制限しています。これを超える場合はハッシュ値に置き換えられます。プレフィックスやコネクター名が長すぎるとユーザー体験が悪化する可能性があります。

この問題に対処するため、バージョン 5.7.1 以降は以下のルールを実装しています。

• プレフィックスなし：動作は従来通りで、長いクライアントID（23バイト超）はハッシュ化されて23バイトに収まります。
• プレフィックスあり：
  • プレフィックスが19バイト以下：プレフィックスは保持され、残りのクライアントID部分は4バイトのハッシュに短縮され、全体が23バイト以内に収まります。
  • プレフィックスが20バイト以上：設定されたプレフィックスをそのまま使用し、クライアントIDの短縮は行いません。

静的クライアントIDの設定

統合で使用するクライアントIDが有限の場合、コネクター設定時に個別ノードに静的クライアントIDセットを割り当てることが可能です。静的クライアントIDを設定するには、EMQX クラスター内の各ノードに対してクライアントIDのリストを提供します。各クライアントIDに対して対応するユーザー名とパスワードを指定できます。これは Azure IoT Hub のように、各デバイス（クライアントID）に固有の認証情報が必要なシナリオで特に有用です。

静的クライアントIDを設定する手順は以下の通りです。

1. 静的 ClientId エントリー セクションで Add ボタンをクリックし、新しい静的クライアントIDエントリーを追加します。必要に応じて複数ノードのエントリーを追加可能です。

2. 各エントリーで以下の項目を入力します。

   • ノード名：クライアントIDを割り当てるノードを指定します。例：emqx@10.0.0.1
   • クライアントID：静的クライアントIDを入力します。例：device1。必要に応じて複数のクライアントIDを追加可能です。
     • ユーザー名：（任意）認証用のユーザー名を指定します。
     • パスワード：（任意）認証用パスワードを指定します。プラットフォームによりデバイス固有のキーやシークレット、証明書などが用いられます（例：Azure IoT Hub の認証キー）。

   設定例：

   ノード名	クライアントID	ユーザー名（任意）	パスワード（任意）
   emqx@10.0.0.1	clientid1	username1	secret1
   	clientid3
   emqx@10.0.0.2	clientid2	username2
   emqx@10.0.0.3	clientid4
   	clientid5

設定ファイルで各ノードごとに static_clientids パラメータを定義することも可能です。

静的クライアントIDを設定した場合、これらのクライアントIDを使った MQTT 接続のみが開始され、pool_size や clientid_prefix など動的クライアントID関連の設定は無効になります。

MQTT ブローカー Sink を使ったルールの作成

このセクションでは、リモート MQTT サービスに転送するデータを指定するルールの作成方法を示します。

1. ダッシュボードの Integration -> Rules ページに移動します。

2. ページ右上の Create ボタンをクリックします。

3. ルールIDに my_rule と入力します。

4. SQL エディター に、t/# トピックからの MQTT メッセージをリモート MQTT サーバーに保存するルールを入力します。SQL は以下の通りです。

   SELECT
     *
   FROM
     "t/#"

   MQTTv5 プロトコルを使用している場合、pub_props フィールドをルール SQL に含めると、パブリッシュプロパティがそのままリモートブローカーに転送されます。上記の SELECT * FROM t/# の例が該当します。

   さらにユーザープロパティを動的に追加したい場合は、ルールの出力に pub_props フィールドを含めてください。例えば、以下のルールはペイロードからキーと値を取り出してユーザープロパティを追加します。

   SELECT
     *,
     map_put(concat('User-Property.', payload.extra_key), payload.extra_value, pub_props) as pub_props
   FROM
     't/#'

5. アクションの追加 で、Action Type ドロップダウンから MQTT Broker を選択します。Action ドロップダウンはデフォルトの Create Action のままにします。この例では新しい Sink を作成し、ルールに追加します。

6. Sink の名前と説明をフォームに入力します。

7. Connector ドロップダウンから先ほど作成した my_mqtt_bridge コネクターを選択します。あるいは、ドロップダウン横の作成ボタンから新規コネクターを作成し、コネクターの作成 の設定パラメータを利用できます。

8. EMQX から外部 MQTT サービスにメッセージをパブリッシュするための Sink 情報を設定します。

   • Topic：外部 MQTT サービスにパブリッシュするトピック。${var} プレースホルダーをサポートします。ここでは pub/${topic} と入力し、元のトピックに pub/ プレフィックスを付けて転送します。例えば元トピックが t/1 の場合、外部 MQTT サービスに転送されるトピックは pub/t/1 になります。
   • QoS：メッセージパブリッシュの QoS。ドロップダウンから 0、1、2、${qos} を選択可能で、他のフィールドから QoS を設定するプレースホルダーも利用できます。ここでは元メッセージの QoS に従う ${qos} を選択します。
   • Retain：メッセージをリテインとしてパブリッシュするかどうか。true、false、${flags.retain} を選択可能で、他のフィールドからリテインフラグを設定するプレースホルダーも利用できます。ここでは元メッセージのリテインフラグに従う ${flags.retain} を選択します。
   • Payload：転送メッセージのペイロード生成テンプレート。デフォルトは空欄でルール出力結果を転送します。ここでは ${payload} と入力し、ペイロードのみを転送します。

9. フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。詳細は フォールバックアクション を参照してください。

10. その他の設定はデフォルトのままにし、Create ボタンをクリックして Sink 作成を完了します。作成後はルール作成ページに戻り、新しい Sink がルールのアクション出力に追加されます。

11. ルール作成ページ下部の Create ボタンをクリックしてルール作成を完了します。

これでルールの作成が完了しました。Integration -> Rules ページで新規作成したルールを確認できます。Actions(Sink) タブをクリックすると新しい MQTT ブローカー Sink が表示されます。

また、Integration -> Flow Designer をクリックするとトポロジーが表示され、トピック t/# のメッセージがルール my_rule によって処理され、リモート MQTT ブローカーに送信される様子を視覚的に確認できます。

MQTT ブローカー Sink を使ったルールのテスト

MQTTX CLI を使って、EMQX の t/# トピックから外部 MQTT サービスの pub/${topic} トピックにメッセージがブリッジされるルールをテストできます。EMQX の t/1 トピックにメッセージをパブリッシュすると、外部 MQTT サービスの pub/t/1 トピックに転送されるはずです。

1. 外部 MQTT サービスで pub/# トピックをサブスクライブします。

   mqttx sub -t pub/# -q 1 -h broker.emqx.io -v

2. MQTTX で t/1 トピックにメッセージをパブリッシュします。

   mqttx pub -t t/1 -m "hello world" -r

3. MQTTX で pub/t/1 トピックをサブスクライブすると、メッセージが受信でき、EMQX から外部 MQTT サービスに正常に転送されたことがわかります。

   [2024-1-31] [16:43:13] › topic: pub/t/1
   payload: hello world

4. ステップ1を繰り返すと、MQTTX で pub/t/1 トピックのリテインメッセージも確認できます。

   [2024-1-31] [16:44:29] › topic: pub/t/1
   payload: hello world
   retain: true

MQTT ブローカー Source を使ったルールの作成

このセクションでは、リモート MQTT サービスからローカル EMQX にデータを転送するルールの作成方法を示します。MQTT Source とメッセージ再パブリッシュアクションを作成し、リモート MQTT サービスのサブスクライブを EMQX に実現し、サブスクライブしたデータを転送します。

MQTT ブローカー Source の作成とルールへの追加

1. ダッシュボードの Integration -> Rules ページに移動します。

2. ページ右上の Create ボタンをクリックします。

3. ルールIDに my_rule_source と入力します。

4. ルールのトリガーソース（データ入力）を設定します。ページ右側の Data Inputs タブで、デフォルトの Message タイプの入力を削除し、Add Input をクリックして MQTT Source を作成します。

5. Add Input ポップアップで、Input Type ドロップダウンから MQTT Broker を選択し、Source ドロップダウンはデフォルトの Create Source のままにします。この例では新しい Source を作成しルールに追加します。

6. Source の名前と説明をフォームに入力します。

7. 先ほど作成した my_mqtt_bridge コネクターをドロップダウンから選択します。あるいは、ドロップダウン横の作成ボタンから新規コネクターを作成し、コネクターの作成 の設定パラメータを利用できます。

8. 外部 MQTT サービスから EMQX へのサブスクライブを完了するため、Source 情報を設定します。

   • Topic：サブスクライブするトピック。+ と # ワイルドカードをサポートします。

     TIP

     EMQX がクラスター運用中、またはコネクターがコネクションプール設定の場合は、重複メッセージを避けるため共有サブスクリプションを使用する必要があります。

     ここでは $share/1/f/# と入力し、f/# にマッチするすべてのメッセージをサブスクライブします。

   • QoS：サブスクライブの QoS。ドロップダウンから 0 または 1 を選択します。

   • No Local：同一コネクターでメッセージをパブリッシュし、そのトピックをこの Source もサブスクライブしている場合に、リモート MQTT ブローカーがメッセージを Source に再転送するのを防止します。デフォルトは無効で、MQTT 5.0 接続時のみ有効です。

   • Retain As Published：アップストリーム MQTT ブローカーが転送するメッセージの元のリテインフラグを保持します。有効にしない場合はアップストリームがリテインフラグをクリアします。デフォルトは有効で、MQTT 5.0 接続時のみ有効です。

9. その他の設定はデフォルトのままにし、Create ボタンをクリックして Source 作成を完了し、ルールのデータ入力に追加します。ルール SQL は以下のように変更されます。

   SELECT
     *
   FROM
     "$bridges/mqtt:my_source"

   ルール SQL は MQTT Source から以下のフィールドを抽出可能で、データ処理のために SQL を調整できます。デフォルトの SQL はこの例に十分です。

   フィールド名	説明
   topic	元のメッセージトピック
   server	接続された Source のサーバーアドレス
   retain	アップストリーム MQTT ブローカーから受信したメッセージのリテインフラグ。Retain As Published 無効時は false
   qos	メッセージの QoS
   pub_props	MQTT 5.0 メッセージプロパティオブジェクト（ユーザープロパティペア、ユーザープロパティ、その他属性を含む）
   pub_props.User-Property-Pairs	キー・バリューのペアを含むユーザープロパティ配列。例：{"key":"foo", "value":"bar"}
   pub_props.User-Property	キー・バリューのペアを含むユーザープロパティオブジェクト。例：{"foo":"bar"}
   pub_props.*	その他のメッセージプロパティのキー・バリュー。例：Content-Type: JSON
   payload	メッセージ内容
   message_received_at	メッセージ受信タイムスタンプ（ミリ秒）
   id	メッセージID
   dup	メッセージが重複かどうか

これで MQTT Source の作成は完了しましたが、サブスクライブしたデータはローカル EMQX に直接パブリッシュされません。次に、Source がサブスクライブしたメッセージをローカル EMQX に転送するためのメッセージ再パブリッシュアクションを作成します。

再パブリッシュアクションの作成

1. ルール作成ページの右側 Action Outputs タブに切り替え、Add Action ボタンをクリックします。Type of Action ドロップダウンから Republish アクションを選択します。

2. メッセージ再パブリッシュの設定を行います。

   • Topic：sub/${topic} と入力し、元のトピックに sub/ プレフィックスを付けて転送します。例えば元トピックが f/1 の場合、EMQX に転送されるトピックは sub/f/1 になります。
   • QoS：0、1、2、${qos} から選択可能。プレースホルダーで他フィールドから QoS を設定することもできます。ここでは元メッセージの QoS に従う ${qos} を選択します。
   • Retain：true または false を選択し、メッセージをリテインとしてパブリッシュするか決定します。プレースホルダーで他フィールドからリテインフラグを設定可能です。この例では false を選択します。
     • データソースが MQTT Source のため、${flags.retain} は使用できません。
     • ${retain} を入力すると MQTT Source が受信したリテインフラグに従います。MQTT 5.0 接続の場合は Source で Retain As Published を有効にしてください。
   • Payload：転送メッセージのペイロード生成に使います。デフォルトは空欄でルール出力結果を転送します。ここでは ${payload} と入力し、ペイロードのみを転送します。

3. Add ボタンをクリックしてアクション作成を完了します。ルール作成ページに戻り、新しいアクションが Action Outputs タブに追加されます。

4. ルール作成ページ下部の Create ボタンをクリックしてルール作成を完了します。

これでルールの作成が完了しました。Integration -> Rules ページで新規作成したルールを確認できます。Source タブをクリックすると新しい MQTT Source が表示されます。

また、Integration -> Flow Designer をクリックするとトポロジーが表示され、MQTT Source からのメッセージが再パブリッシュアクションを通じて sub/${topic} に転送される様子を視覚的に確認できます。

MQTT ブローカー Source を使ったルールのテスト

MQTTX CLI を使って、外部 MQTT サービスの f/# トピックから EMQX の sub/${topic} トピックにメッセージがブリッジされるルールをテストできます。外部 MQTT サービスの f/1 トピックにメッセージをパブリッシュすると、EMQX の sub/f/1 トピックに転送されるはずです。

1. EMQX の sub/# トピックをサブスクライブします。

   mqttx sub -t sub/# -q 1 -v

2. MQTTX で外部 MQTT サービスの f/1 トピックにメッセージをパブリッシュします。

   mqttx pub -t f/1 -m "I'm from broker.emqx.io" -r -h broker.emqx.io

3. MQTTX で sub/f/1 トピックにパブリッシュされたメッセージを確認できれば、外部 MQTT サービスから EMQX への正常な転送が確認できます。

   [2024-1-31] [16:49:22] › topic: sub/f/1
   payload: I'm from broker.emqx.io