EMQX Cloud

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

Bridge MQTT Data into RocketMQ ​

EMQX CloudはRocketMQへのデータブリッジをサポートしており、MQTTメッセージやクライアントイベントをRocketMQに転送できます。例えば、RocketMQを使用してデバイスからのセンサーデータやログデータを収集することが可能です。

本ページでは、EMQX CloudとRocketMQ間のデータ統合の詳細な概要と、データ統合の作成および検証に関する実践的な手順を提供します。

How It Works ​

RocketMQデータ統合は、EMQX Cloudに標準搭載された機能であり、EMQX Cloudのリアルタイムデータキャプチャおよび送信能力とRocketMQの強力なメッセージキュー処理能力を組み合わせています。組み込みのルールエンジンコンポーネントにより、EMQX CloudからRocketMQへのデータ取り込みプロセスを簡素化し、複雑なコーディングを不要にします。

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

EMQX Cloud-RocketMQ Integration

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

  1. メッセージのパブリッシュと受信：産業用IoTデバイスはMQTTプロトコルを介してEMQX Cloudに正常に接続し、リアルタイムのMQTTデータをEMQX Cloudにパブリッシュします。EMQX Cloudがこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
  2. メッセージデータの処理：メッセージが到着するとルールエンジンを通過し、EMQX Cloudで定義されたルールによって処理されます。ルールは事前定義された条件に基づき、RocketMQにルーティングすべきメッセージを決定します。ペイロード変換が指定されている場合は、データ形式の変換、特定情報のフィルタリング、ペイロードのコンテキスト付加などの変換が適用されます。
  3. RocketMQへのデータ取り込み：ルールによるメッセージ処理が完了すると、メッセージをRocketMQに転送するアクションがトリガーされます。処理済みデータはシームレスにRocketMQに書き込まれます。
  4. データの保存と活用：データがRocketMQに保存されることで、企業はそのクエリ機能を活用して様々なユースケースに対応できます。例えば金融業界では、RocketMQを信頼性の高い高性能メッセージキューとして使用し、決済端末や取引システムからのデータを管理します。メッセージをデータ分析や規制プラットフォームに接続し、リスク管理、不正検知・防止、規制遵守などの要件を満たせます。

Features and Benefits ​

RocketMQとのデータ統合は、以下の特徴と利点をもたらします。

  • 信頼性の高いIoTデータメッセージ配信：EMQX CloudはMQTTメッセージをバッチ処理で確実にRocketMQに送信でき、IoTデバイスとRocketMQおよびアプリケーションシステムの統合を実現します。
  • MQTTメッセージ変換：ルールエンジンを使用して、EMQX CloudはMQTTメッセージのフィルタリングや変換を行えます。メッセージはデータ抽出、フィルタリング、拡充、変換を経てRocketMQに送信されます。
  • クラウドネイティブな弾力的スケーリング：EMQX CloudとRocketMQはともにクラウドネイティブアーキテクチャ上に構築されており、Kubernetes（K8s）との親和性やクラウドネイティブエコシステムとの統合を備えています。ビジネスの急速な成長に対応して無限かつ弾力的にスケール可能です。
  • 柔軟なトピックマッピング：RocketMQデータ統合はMQTTトピックとRocketMQトピックの柔軟なマッピングをサポートし、RocketMQメッセージ内のキー（Key）や値（Value）の設定を簡単に行えます。
  • 高スループットシナリオでの処理能力：RocketMQデータ統合は同期・非同期の書き込みモードをサポートし、シナリオに応じてレイテンシとスループットのバランスを柔軟に調整できます。

Before You Start ​

このセクションでは、RocketMQデータ統合の作成を開始する前に必要な準備、特にRocketMQサーバーのセットアップ方法について説明します。

Prerequisites ​

Set up Network ​

データ統合を構成する前に、EMQX Cloudのデプロイメントを作成し、EMQX Cloudと対象サービス間のネットワーク接続を確立していることを確認してください。

  • Dedicated Flexデプロイメントの場合：

    EMQX CloudのVPCと対象サービスのVPC間でVPCピアリング接続を作成します。ピアリング接続が確立されると、EMQX Cloudは対象サービスのプライベートIPアドレスを介してアクセス可能になります。

    パブリックIP経由でのアクセスが必要な場合は、NATゲートウェイを構成してアウトバウンド接続を有効にしてください。

  • BYOC（Bring Your Own Cloud）デプロイメントの場合：

    BYOCデプロイメントが稼働しているVPCと対象サービスをホストするVPC間でVPCピアリング接続を作成します。ピアリングが確立されると、対象サービスのプライベートIPアドレスを介してアクセス可能になります。

    対象サービスにパブリックIP経由でアクセスする必要がある場合は、クラウドプロバイダーのコンソールを使用してBYOC VPCにNATゲートウェイを構成してください。

Install RocketMQ ​

  1. RocketMQをセットアップするためのdocker-composeファイルrocketmq.yamlを用意します。
yaml
version: '3.3'

services:
  mqnamesrv:
    image: apache/rocketmq:4.9.4
    container_name: rocketmq_namesrv
    ports:
      - 9876:9876
    volumes:
      - ./rocketmq/logs:/opt/logs
      - ./rocketmq/store:/opt/store
    command: ./mqnamesrv

  mqbroker:
    image: apache/rocketmq:4.9.4
    container_name: rocketmq_broker
    ports:
      - 10909:10909
      - 10911:10911
    volumes:
      - ./rocketmq/logs:/opt/logs
      - ./rocketmq/store:/opt/store
      - ./rocketmq/conf/broker.conf:/etc/rocketmq/broker.conf
    environment:
      NAMESRV_ADDR: 'rocketmq_namesrv:9876'
      JAVA_OPTS: ' -Duser.home=/opt'
      JAVA_OPT_EXT: '-server -Xms1024m -Xmx1024m -Xmn1024m'
    command: ./mqbroker -c /etc/rocketmq/broker.conf
    depends_on:
      - mqnamesrv
  1. RocketMQの実行に必要なフォルダと設定を準備します。
bash
mkdir rocketmq
mkdir rocketmq/logs
mkdir rocketmq/store
mkdir rocketmq/conf
  1. 以下の内容をrocketmq/conf/broker.confに保存します。
bash
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0

brokerIP1=change me to your real IP address

defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true

listenPort=10911
deleteWhen=04

fileReservedTime=120
mapedFileSizeCommitLog=1073741824
mapedFileSizeConsumeQueue=300000
diskMaxUsedSpaceRatio=100
maxMessageSize=65536

brokerRole=ASYNC_MASTER

flushDiskType=ASYNC_FLUSH
  1. サーバーを起動します。
bash
docker-compose -f rocketmq.yaml up
  1. コンシューマーを起動します。
docker run --rm -e NAMESRV_ADDR=host.docker.internal:9876 apache/rocketmq:4.9.4 ./tools.sh org.apache.rocketmq.example.quickstart.Consumer

TIP

Linux環境では、host.docker.internalを実際のIPアドレスに変更してください。

Create a Connector ​

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

  1. デプロイメントに移動し、左側ナビゲーションメニューからデータ統合をクリックします。初めてコネクターを作成する場合は、データ転送カテゴリの下にあるRocketMQを選択します。すでにコネクターを作成している場合は、新しいコネクターを選択し、続いてデータ転送カテゴリの下のRocketMQを選択します。

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

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

    • Servers：接続したいRocketMQサーバーのアドレスを入力します（例：localhost）。ホストエントリはHost[:Port]の形式です。[:Port]を指定しない場合、RocketMQのデフォルトポート9876が使用されます。
    • RocketMQ Namespace：EMQX 6.1.5以降のデプロイメントでは、RocketMQサービスに設定されたネームスペースを入力します。Alibaba Cloud RocketMQの場合はインスタンスIDを入力します。RocketMQサービスがネームスペースを使用していない場合は空欄のままにします。
    • AccessKey：RocketMQサーバーのAccessKey。
    • SecretKey：RocketMQサーバーのSecretKey。
    • Security Token：RocketMQサーバーのセキュリティトークン。
    • ビジネスニーズに応じて詳細設定を行います（任意）。

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

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

これで、このコネクターを基にデータブリッジルールを作成できます。

Create a Rule ​

このセクションでは、RocketMQルールを作成し、EMQX Cloudコンソールを通じてルールにアクションを追加する方法を説明します。

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

  2. 利用したい機能に基づいてSQLエディターでルールを設定します。ここでは、クライアントがtemp_hum/emqxトピックに温度と湿度のメッセージを送信した際にエンジンをトリガーするSQL例を示します。

sql
  SELECT

   timestamp as up_timestamp,
   clientid as client_id,
   payload.temp as temp,
   payload.hum as hum

  FROM

   "temp_hum/emqx"

TIP

初心者の方は、SQL例をクリックし、試してみるでSQLルールを学習・テストできます。

  1. 次へをクリックしてアクションを追加します。

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

  3. EMQX CloudからRocketMQサービスにメッセージをパブリッシュするための情報を設定します：

    • RocketMQトピック：TopicTest

    • メッセージテンプレート：テンプレートのデフォルト値は空です。空の場合はメッセージ全体がRocketMQに保存されます。テンプレートはプレースホルダーを含む有効な文字列で、例：

      {"up_timestamp": ${up_timestamp}, "client_id": ${client_id}, "temp": ${temp}, "hum": ${hum}}

  4. 詳細設定を展開し、同期/非同期モード、キューやバッチ、その他パラメータを適宜設定します（任意）。

  5. 確認ボタンをクリックしてルール作成を完了します。

  6. 新規ルール作成成功のポップアップでルールに戻るをクリックし、データ統合設定の一連の流れを完了します。

Test the Rule ​

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

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

    • トピック：temp_hum/emqx

    • クライアントID：test_client

    • ペイロード：

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

  2. RocketMQコンシューマーのウィンドウに以下のような出力が表示されます。

    bash
    ConsumeMessageThread_please_rename_unique_group_name_4_1 Receive New Messages: [MessageExt [brokerName=broker-a, queueId=0, storeSize=208, queueOffset=0, sysFlag=0, bornTimestamp=1711354009076, bornHost=/121.43.165.169:48850, storeTimestamp=1711354009085, storeHost=/118.178.124.161:10911, msgId=76B27CA100002A9F000000000000058D, commitLogOffset=1421, bodyCRC=1137462344, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, CONSUME_START_TIME=1711354066863, CLUSTER=DefaultCluster}, body=[123, 34, 117, 112, 95, 116, 105, 109, 101, 115, 116, 97, 109, 112, 34, 58, 49, 55, 49, 49, 51, 53, 52, 48, 48, 57, 48, 53, 54, 44, 34, 116, 101, 109, 112, 34, 58, 34, 50, 55, 46, 53, 34, 44, 34, 104, 117, 109, 34, 58, 34, 52, 49, 46, 56, 34, 44, 34, 99, 108, 105, 101, 110, 116, 95, 105, 100, 34, 58, 34, 109, 113, 116, 116, 120, 95, 97, 50, 97, 99, 102, 100, 49, 57, 34, 125], transactionId='null'}]]