EMQX Platform

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ

RabbitMQへのMQTTデータ取り込み ​

RabbitMQは、Advanced Message Queuing Protocol（AMQP）を実装した広く利用されているオープンソースのメッセージブローカーです。分散システム間のメッセージングにおいて堅牢でスケーラブルなプラットフォームを提供します。EMQXプラットフォームはRabbitMQとの統合をサポートしており、MQTTメッセージやイベントをRabbitMQに転送できます。また、RabbitMQサーバーからデータを取得し、EMQXプラットフォームの特定のトピックにパブリッシュすることも可能で、RabbitMQからMQTTへのメッセージ配信を実現します。

本ページでは、EMQXプラットフォームとRabbitMQ間のデータ統合について詳細に解説し、データ統合の作成および検証手順を実践的に説明します。

動作概要 ​

RabbitMQデータ統合は、EMQXプラットフォームに標準搭載された機能であり、MQTTベースのIoTデータとRabbitMQの強力なメッセージキュー処理機能を橋渡しします。組み込みのルールエンジンコンポーネントを活用することで、EMQXプラットフォームからRabbitMQへのデータ取り込みをコード不要で簡単に実現できます。

以下の図は、EMQXプラットフォームとRabbitMQ間の典型的なデータ統合アーキテクチャを示しています。

emqx-integration-rabbitmq

MQTTデータをRabbitMQに取り込む流れは以下の通りです：

  1. メッセージのパブリッシュと受信：産業用IoTデバイスはMQTTプロトコルを通じてEMQXプラットフォームに正常に接続し、リアルタイムMQTTデータをパブリッシュします。EMQXプラットフォームがメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
  2. メッセージデータの処理：メッセージ到着後、ルールエンジンを通過し、EMQXプラットフォームで定義されたルールにより処理されます。ルールは事前定義された条件に基づき、RabbitMQにルーティングすべきメッセージを判別します。ペイロード変換が指定されている場合は、データ形式の変換、特定情報のフィルタリング、追加コンテキストによるペイロードの強化などが適用されます。
  3. RabbitMQへのメッセージ取り込み：ルールによる処理完了後、メッセージをRabbitMQに転送するアクションがトリガーされます。処理済みメッセージはシームレスにRabbitMQに書き込まれます。
  4. データの永続化と活用：RabbitMQはメッセージをキューに格納し、適切なコンシューマーに配信します。メッセージは他のアプリケーションやサービスによって消費され、データ分析、可視化、保存などのさらなる処理に利用されます。

特長と利点 ​

RabbitMQとのデータ統合は、以下の特長とメリットをもたらします：

  • 信頼性の高いIoTデータメッセージ配信：EMQXプラットフォームはデバイスからクラウドへの信頼性の高い接続とメッセージ配信を保証し、RabbitMQはメッセージの永続化と異なるサービス間の信頼性ある配信を担い、全体のデータ信頼性を確保します。
  • MQTTメッセージの変換：ルールエンジンを用いて、EMQXプラットフォームはMQTTメッセージのフィルタリングや変換を実施可能です。データ抽出、フィルタリング、強化、変換を経てRabbitMQに送信されます。
  • 柔軟なメッセージマッピング：RabbitMQデータ統合はMQTTトピックとRabbitMQのルーティングキーおよびエクスチェンジの柔軟なマッピングをサポートし、MQTTとRabbitMQ間のシームレスな連携を実現します。
  • 高可用性とクラスター対応：EMQXプラットフォームとRabbitMQは共に高可用性のメッセージブローカークラスター構築をサポートし、ノード障害時もサービス継続を保証します。クラスター機能により優れたスケーラビリティも提供されます。
  • 高スループットシナリオでの処理能力：RabbitMQデータ統合は同期および非同期書き込みモードをサポートし、レイテンシとスループットのバランスを柔軟に調整可能です。

はじめる前に ​

このセクションでは、RabbitMQデータ統合の作成を始める前に必要な準備、RabbitMQサーバーの作成方法およびテスト用のエクスチェンジとキューの作成方法を説明します。

前提条件 ​

  • データ統合の知識
  • EMQXプラットフォームのデータ統合ルールの知識
  • UNIXターミナルとコマンドの基本知識

ネットワーク設定 ​

EMQX Platform上でデプロイメント（EMQXクラスター）を作成し、ネットワークを構成する必要があります。

  • 専用デプロイメントユーザーの場合：まず、VPCピアリング接続を作成してください。ピアリング接続が確立された後、内部ネットワークのIPを介してプラットフォームコンソールにログインし、対象のコネクターにアクセスできます。あるいは、NATゲートウェイを設定して、パブリックIP経由で対象のコネクターにアクセスすることも可能です。
  • BYOC（Bring Your Own Cloud）デプロイメントユーザーの場合：BYOCがデプロイされているVPCと対象コネクターが存在するVPC間でピアリング接続を確立してください。ピアリング接続作成後、内部ネットワークのIPを介して対象コネクターにアクセスできます。パブリックIPアドレス経由でリソースにアクセスする必要がある場合は、パブリッククラウドコンソールでBYOCがデプロイされているVPCに対してNATゲートウェイを構成してください。

RabbitMQサーバーの起動 ​

ここでは、Dockerを用いてRabbitMQサーバーを起動する方法を紹介します。

管理プラグインを有効にしたRabbitMQサーバーを起動するには、以下のコマンドを実行してください。管理プラグインにより、WebインターフェースでRabbitMQの状態を確認できます。

bash
docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.11-management

Docker HubのRabbitMQのDockerイメージに関する詳細情報もご参照ください。

メッセージ受信用のエクスチェンジとキューの作成 ​

RabbitMQサーバー起動後、RabbitMQ管理Webインターフェースを使って、EMQXプラットフォームから転送されるメッセージを受信するためのテスト用エクスチェンジとキューを作成します。既にテスト用のエクスチェンジとキューがある場合は、このセクションはスキップ可能です。

  1. Webブラウザで http://{ip address}:15672/ にアクセスし、RabbitMQ管理Webインターフェースを開きます。ログイン画面で以下のデフォルト認証情報を入力し、Loginをクリックします。

    • Username：guest
    • Password：guest

  2. 上部メニューのExchangesタブをクリックします。Add a new exchangeを展開し、以下の情報を入力します：

    • Name：test_exchangeと入力
    • Type：ドロップダウンリストからdirectを選択
    • Durability：Durableを選択し、エクスチェンジを永続化（RabbitMQ再起動後も存在）
    • Auto delete：No
    • Internal：No
    • Arguments：空欄のまま

  3. Add exchangeボタンをクリックしてテスト用エクスチェンジを作成します。

  4. 上部メニューのQueuesタブをクリックします。Add a new queueを展開し、以下の情報を入力します：

    • Type：Default for virtual host
    • Name：test_queueと入力
    • Durability：Durableを選択し、キューを永続化
    • Arguments：空欄のまま

  5. Add queueボタンをクリックしてテスト用キューを作成します。新しいtest_queueがAll queuesセクションに表示されます。

  6. キュー名のtest_queueをクリックして詳細ページを開きます。Bindingsを展開し、Add binding to this queueセクションに以下を入力します：

    • From exchange：test_exchangeと入力
    • Routing key：test_routing_keyと入力
    • Arguments：空欄のまま

  7. Bindボタンをクリックして、test_queueを指定したルーティングキーでtest_exchangeにバインドします。

コネクターの作成 ​

データ統合ルールを作成する前に、RabbitMQサーバーにアクセスするためのRabbitMQコネクターを作成する必要があります。

  1. デプロイメントに移動し、左ナビゲーションメニューからData Integrationをクリックします。初めてコネクターを作成する場合は、Data Forwardカテゴリの下にあるRabbitMQを選択します。既にコネクターを作成済みの場合は、New Connectorを選択し、続けてData ForwardカテゴリのRabbitMQを選択します。

  2. Connector Name：システムが自動的にコネクター名を生成します。

  3. 接続情報を入力します：

    • Server：RabbitMQサーバーがローカルの場合はlocalhost、リモートの場合は実際のホスト名またはIPアドレスを入力
    • Port：通常は5672、異なる場合は実際のポート番号を入力
    • Username：guest
    • Password：guest
    • Virtual Host：RabbitMQの仮想ホスト。デフォルトは/
    • Enable TLS：暗号化接続を確立したい場合はトグルスイッチをオンにする
    • ビジネスニーズに応じて詳細設定を構成（任意）

  4. Testボタンをクリックします。RabbitMQサービスにアクセス可能であれば、connector availableのプロンプトが返されます。

  5. Newボタンをクリックして作成を完了します。

プロデューサールールの作成 ​

このセクションでは、RabbitMQプロデューサールールを作成し、EMQXプラットフォームコンソールからルールにアクションを追加する方法を示します。

  1. ルールエリアでNew Ruleをクリックするか、作成したコネクターのActions列にある新規ルールアイコンをクリックします。

  2. 使用する機能に基づき、SQL Editorでルールを設定します。ここでは、クライアントがtemp_hum/emqxトピックに温度と湿度のメッセージを送信した際にエンジンをトリガーするSQLを記述します。

    sql
     SELECT
  timestamp, clientid, payload
 FROM
   "temp_hum/emqx"

    TIP

    初心者の方は、SQL Examplesをクリックし、Enable Testを有効にしてSQLルールの学習とテストを行うことを推奨します。

  3. Nextをクリックしてアクションを追加します。

  4. Connectorのドロップダウンボックスから先ほど作成したコネクターを選択します。

  5. EMQXプラットフォームからRabbitMQサービスにメッセージをパブリッシュするための情報を設定します：

    • Exchange：前に作成したtest_exchangeを入力。メッセージはこのエクスチェンジにパブリッシュされます。

    • Routing Key：前に作成したtest_routing_keyを入力。RabbitMQでメッセージパブリッシュ時に使用されるルーティングキーです。

    • Message Delivery Modeのドロップダウンからnon_persistentまたはpersistentを選択：

      • non_persistent（デフォルト）：メッセージはディスクに永続化されず、RabbitMQの再起動やクラッシュ時に失われる可能性があります。

      • persistent：メッセージはディスクに永続化され、RabbitMQの再起動やクラッシュ時にも耐久性があります。

      TIP

      RabbitMQが再起動した場合にメッセージを失わないように、キューとエクスチェンジも永続化（Durable）に設定する必要があります。詳細はRabbitMQのドキュメントをご参照ください。

    • Payload Template：デフォルトは空文字列で、メッセージペイロードはJSON形式のテキストとしてRabbitMQにそのまま転送されます。

      プレースホルダーを使ってカスタムのメッセージペイロードフォーマットを定義することも可能です。例えば、MQTTメッセージのペイロードとタイムスタンプをRabbitMQメッセージに含めたい場合は、以下のテンプレートを使用します：

      json
      {"payload": "${payload}", "timestamp": ${timestamp}}

      このテンプレートは、転送されるMQTTメッセージのペイロードとタイムスタンプを含むJSON形式のメッセージを生成します。${payload}と${timestamp}はプレースホルダーで、実際のメッセージの値に置き換えられます。

    • Wait for Publish Confirmations：デフォルトで有効。メッセージがRabbitMQに正常にパブリッシュされたことを確認します。

      TIP

      このオプションを有効にすると、RabbitMQブローカーはメッセージの受信をアック（ACK）し、成功したパブリッシュとして扱うため、メッセージ配信の信頼性が向上します。

  6. Advanced Settingsを展開し、同期/非同期モード、キューやバッチ処理などのパラメータを必要に応じて設定します（任意）。

  7. Confirmボタンをクリックしてルール作成を完了します。

  8. Successful new ruleのポップアップでBack to Rulesをクリックし、データ統合設定の一連の作業を完了します。

プロデューサールールのテスト ​

温度と湿度のデータ報告をシミュレートするために、MQTTXの使用を推奨しますが、他のクライアントでも構いません。

  1. MQTTXを使ってデプロイメントに接続し、以下のトピックにメッセージを送信します。

    • トピック：temp_hum/emqx

    • クライアントID：test_client

    • ペイロード：

      json
      {
  "temp": "27.5",
  "hum": "41.8"
}

  2. アクションとルールが正常に作成されていれば、指定したルーティングキーでRabbitMQサーバーの指定エクスチェンジにメッセージがパブリッシュされているはずです。RabbitMQ管理コンソール（http://{ip address}:15672）にアクセスし、Queuesセクションに移動してください。

  3. メッセージが適切なキューにルーティングされていることを確認します。キューをクリックし、**Get Message(s)**ボタンを押して詳細なメッセージ内容を確認します。

    • ペイロード例：

      json
      {"payload":
 "{
 "temp": "27.5",
 "hum": "41.8"
  }",
  "timestamp": 1711333401673
}

RabbitMQソースルールの作成 ​

このセクションでは、RabbitMQソースルールを作成し、RabbitMQソースから消費したメッセージをEMQXプラットフォームに転送してトピックt/1にパブリッシュするリパブリッシュアクションを追加する方法を示します。

  1. ルールエリアでNew Ruleをクリックするか、作成したコネクターのActions列にある新規ルールアイコンをクリックします。

  2. ルールIDにmy_rule_sourceを入力します。

  3. ルールをトリガーするソース（データ入力）を設定します。ページ右側の**Actions (Source)**タブをクリックし、**New Action (Source)**をクリックしてRabbitMQソースを作成します。

  4. スライドパネルでソースタイプに**RabbitMQ (Source)**を選択し、Nextをクリックして設定ステップに進みます。

  5. Connectorのドロップダウンから、先に作成したmy-rabbitmqコネクターを選択します。ドロップダウン横の作成ボタンをクリックすると、ポップアップで新規コネクターを素早く作成可能です。必要な設定パラメータはコネクターの作成を参照してください。

  6. RabbitMQからEMQXプラットフォームへメッセージを消費するためのソース情報を設定します：

    • Queue：RabbitMQで先に作成したキュー名message-sendを入力
    • No Ack：状況に応じて選択。no_ackモードでRabbitMQからメッセージを消費するかどうかを指定します。no_ackを有効にすると、RabbitMQはメッセージをコンシューマーが正常に処理したかに関わらず即座にキューから削除します。
    • Wait for Publish Confirmations：メッセージパブリッシャーのアックを使用する際に、RabbitMQからの確認を待つかどうかを指定します。

  7. 高度な設定（任意）：デフォルト値を使用します。

  8. Confirmボタンをクリックしてソース作成を完了し、ルールのデータ入力に追加します。同時にルールSQLが以下のように変更されます。

    sql
    SELECT
*
FROM
"$bridges/rabbitmq:my-rabbitmq-source"

    RabbitMQソースから以下のフィールドにアクセス可能であり、SQLを調整してデータ処理を行えます。ここではデフォルトSQLを使用します。

    フィールド名説明
    payloadRabbitMQメッセージの内容
    eventイベントトピック。形式は$bridges/rabbitmq:<source name>
    metadataルールID情報
    timestampメッセージがEMQXに到着したタイムスタンプ
    nodeメッセージが到着したEMQXノード名
    queueメッセージが消費されたキュー名
    exchangeメッセージがルーティングされたエクスチェンジ名
    routing_keyエクスチェンジからキューへメッセージをルーティングする際のルーティングキー

  9. Nextをクリックし、出力アクションを作成します。

  10. 新しい出力アクションでRepublishを選択します。

  11. メッセージのリパブリッシュ設定を入力します：

    • Topic：MQTTにパブリッシュするトピック。ここではt/1を入力。
    • QoS：0、1、2、または${qos}を選択。${qos}を選ぶと元のメッセージのQoSに従います。
    • Retain：trueまたはfalseを選択。メッセージをリテインメッセージとしてパブリッシュするかを決定。プレースホルダーも使用可能。ここではfalseを選択。
    • Payload：転送するメッセージペイロードのテンプレート。空欄の場合はルールの出力結果をそのまま転送。ここでは${payload}と入力し、ペイロードのみを転送。
    • MQTT 5.0 Message Properties：デフォルトで無効。詳細はリパブリッシュアクションの追加を参照。

  12. その他の設定はデフォルトのままにし、Confirmボタンをクリックして出力アクションの作成を完了します。

作成成功後、New Ruleページに戻ります。Rulesリストに新規作成したルールが表示されます。リパブリッシュアクションは現在**Actions (Source)**には表示されません。必要に応じてルール編集ボタンをクリックすると、ルール設定の下部にリパブリッシュ出力アクションが確認できます。

RabbitMQソースルールのテスト ​

  1. MQTTX CLIを使用してトピックt/1をサブスクライブします：

    bash
    mqttx sub -t t/1

  2. 以下のコマンドでRabbitMQにメッセージをパブリッシュできます：

    bash
    rabbitmqadmin --username=guest --password=guest \
     publish routing_key=message-send \
     payload="{ \"msg\": \"Hello EMQX\"}"
    • publishはメッセージをパブリッシュするコマンドです。
    • routing_key=message-sendオプションはメッセージのルーティングキーを設定します。この例ではキュー名をルーティングキーとして使用しています。
    • payload="{ \"msg\": \"Hello EMQX\"}"オプションはメッセージの内容を設定します。

    または、RabbitMQ管理インターフェースからもメッセージをパブリッシュ可能です：

    1. 上部メニューのQueuesタブをクリックします。
    2. Name列のmessage-sendをクリックして詳細ページを開きます。
    3. Publish messageを展開し、Payloadボックスに"Hello EMQX"と入力して、Publish messageボタンをクリックします。

  3. MQTTXで以下のような出力が確認できます：

    bash
    [2024-2-23] [16:59:28] › payload: {"payload":{"msg":"Hello EMQX"},"event":"$bridges/rabbitmq:my-rabbitmq-source","metadata":{"rule_id":"rule_0ly1"},"timestamp":1708678768449,"node":"emqx@127.0.0.1"}