EMQX Enterprise

日本語

简体中文
English

日本語

简体中文
English

Appearance

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

EMQXでNode-REDを使う ​

Node-REDは、ハードウェアデバイス、API、オンラインサービスをブラウザベースのエディターで接続するためのフローベースプログラミングツールです。視覚的なノードベースのインターフェースを用いて、あらかじめ用意されたノードをつなげてデータフローを作成します。Node-REDは組み込みのmqtt-in（サブスクライブ）およびmqtt-out（パブリッシュ）ノードを通じてMQTTをネイティブにサポートしており、EMQXからのIoTデータ処理に広く利用されています。

本ページでは、Node-REDのインストール方法、EMQXへの接続方法、およびMQTTメッセージを解析、フィルタリング、変換するデータ処理パイプラインの構築方法を説明します。

前提条件 ​

  • Node.js 18 LTS または 20 LTS（NPMインストール用）
  • EMQXのデプロイ環境、またはテスト用にEMQXパブリックブローカーの利用
  • テストメッセージ送信用のMQTTXなどのMQTTクライアント

Node-REDのインストール ​

NPM経由の場合:

bash
npm install -g --unsafe-perm node-red

続いてNode-REDを起動します:

bash
node-red

Docker経由の場合:

bash
docker run -it -p 1880:1880 --name mynodered nodered/node-red

起動後、ブラウザでhttp://127.0.0.1:1880にアクセスするとNode-REDエディターが開きます。

Node-REDエディターへのアクセス

Raspberry Piやクラウドデプロイなどの他のインストールオプションについては、Node-REDドキュメントをご参照ください。

MQTTブローカーの設定 ​

Node-REDを接続するためのMQTTブローカーが必要です。本ガイドではMQTT 3.1、3.1.1、5.0をサポートするEMQXを使用します。

EMQXパブリックブローカー（テスト用） ​

独自のブローカーをデプロイせずに手早くテストしたい場合は、EMQXパブリックブローカーを利用できます。

パラメータ値
ブローカーアドレスbroker.emqx.io
TCPポート1883
SSL/TLSポート8883
WebSocketポート8083
セキュアWebSocketポート8084

パブリックブローカーはテストおよびデモ目的のみの利用を想定しています。

EMQX Enterpriseデプロイメント ​

本番環境では、独自のEMQX EnterpriseデプロイメントにNode-REDを接続し、環境に応じたブローカーアドレス、ポート、認証情報を使用してください。

一般的な構成例は以下の通りです：

  • カスタムブローカーホスト名またはIPアドレス
  • ユーザー名／パスワード認証または相互TLS認証
  • トピックに適用されるアクセス制御ルール（ACL）

Node-REDのブローカー接続設定時には、EMQX Enterpriseのリスナーおよび認証設定を参照してください。

自己管理のEMQX Enterpriseデプロイメントに加え、完全マネージドMQTTサービスであるEMQX Cloud（サーバレスまたは専用）への接続も可能です。EMQX Cloudが提供するブローカーアドレス、ポート、認証情報を使用してください。

基本的なMQTTフローの構築 ​

以下の手順で、1つのトピックをサブスクライブし、受信したメッセージを別のトピックに転送する最小限のフローを作成します。

ステップ1: MQTTサブスクライブノードを追加 ​

  1. Node-REDエディターの左パレットからmqtt-inノードをキャンバスにドラッグします。

  2. ノードをダブルクリックしてプロパティを開きます。

  3. Server欄の横にある鉛筆アイコンをクリックして新しいブローカー接続を作成します。

  4. Serverアドレスにbroker.emqx.ioを入力し、Addをクリックします。

    サブスクライブノードの追加

  5. Topicにtest/node_red/inを設定します。

  6. 必要に応じてQoSレベルを設定し、Doneをクリックします。

    トピックのサブスクライブ設定

ステップ2: MQTTパブリッシュノードを追加 ​

  1. mqtt-outノードをキャンバスにドラッグします。

  2. ノードをダブルクリックしてプロパティを開きます。

  3. ステップ1で設定したブローカーをServerドロップダウンから選択します。

  4. Topicにtest/node_red/outを設定します。

  5. 必要に応じてQoSとRetainを設定し、Doneをクリックします。

    パブリッシュノードの設定

ステップ3: 接続とデプロイ ​

  1. mqtt-inノードの出力ポートからmqtt-outノードの入力ポートへワイヤーを引きます。

  2. 右上のDeployボタンをクリックします。

  3. 両ノードが緑色のconnectedステータスを表示していることを確認します。

これでtest/node_red/inで受信したすべてのメッセージがtest/node_red/outに転送されるフローが完成しました。

ノードの接続

高度なデータ処理パイプラインの構築 ​

Node-REDの真価は複数のノードを連結してデータをフィルタリングや変換し、再パブリッシュできる点にあります。以下の例では、

  1. MQTT経由でJSON形式のセンサーデータを受信し、
  2. 生のペイロードをJavaScriptオブジェクトにパースし、
  3. 重複する温度データをフィルタリングし、
  4. 結果を整形して再パブリッシュします。

フロー全体は次の通りです：mqtt-in -> json -> rbe -> template -> mqtt-out

ステップ1: JSONノードを追加 ​

  1. パレットからjsonノードをキャンバスにドラッグします。

  2. ダブルクリックして設定を開き、ActionをAlways Convert to JavaScript Objectに設定します。

  3. Doneをクリックします。

  4. mqtt-inノードの出力をjsonノードの入力に接続します。

これにより、受信したペイロードがJavaScriptオブジェクトに変換され、下流のノードがmsg.payload.temperatureなどの個別フィールドにアクセス可能になります。

jsonノードの接続

ステップ2: フィルターノードを追加 ​

  1. rbe（report by exception）ノードをキャンバスにドラッグします。

  2. ダブルクリックして設定を開きます：

    • Modeをblock unless value changesに設定
    • Propertyをmsg.payload.temperatureに設定

  3. Doneをクリックします。

  4. jsonノードの出力をrbeノードの入力に接続します。

このフィルターノードは、前回のメッセージと温度値が変わらない場合にメッセージをブロックし、同一の繰り返しデータによる不要なトラフィックを削減します。

フィルターノードの追加

ステップ3: テンプレートノードを追加 ​

  1. templateノードをキャンバスにドラッグします。

  2. ダブルクリックして設定を開き、Mustache構文を用いて出力フォーマットを入力します。例：

    {"temperature": {{payload.temperature}}, "humidity": {{payload.humidity}}}

  3. Doneをクリックします。

  4. rbeノードの出力をtemplateノードの入力に接続します。

テンプレートノードの追加

ステップ4: 出力ノードを接続してデプロイ ​

  1. templateノードの出力をmqtt-outノードの入力に接続します。

  2. Deployをクリックします。

  3. すべてのノードが緑色のconnectedステータスを表示していることを確認します。

フィルタリングしたデータを再フォーマットせずにそのままパブリッシュしたい場合は、templateノードを省略してrbeノードを直接mqtt-outに接続してください。

高度なノードの接続

フローのテスト ​

MQTTXや任意のMQTTクライアントを使ってパイプラインをテストします：

  1. test/node_red/outをサブスクライブして処理結果を確認します。

  2. test/node_red/inにJSONペイロードのテストメッセージをパブリッシュします。例：

    json
    {"temperature": 25, "humidity": 60}

  3. 出力トピックにメッセージが現れることを確認します。

  4. 同じメッセージを再度パブリッシュすると、rbeフィルターが重複を抑制し、出力は表示されません。

  5. 温度値を変更してパブリッシュします：

    json
    {"temperature": 26, "humidity": 60}

  6. このメッセージはフィルターを通過し、出力トピックに表示されることを確認します。

フローのテスト

トラブルシューティング ​

ノードが「disconnected」ステータスを表示する ​

説明

  • デプロイ後、mqtt-inまたはmqtt-outノードが赤色のdisconnectedインジケーターを表示する。

考えられる原因

  • ブローカーアドレスまたはポートの誤り
  • ネットワークファイアウォールによるポート1883または8883のブロック
  • ブローカーが起動していない

対処方法

  • ノードをダブルクリックし、Server横の鉛筆アイコンをクリックしてブローカーアドレスとポートを確認する。
  • MQTTXなど別のMQTTクライアントでブローカーへの基本接続をテストする。
  • TLS利用時は正しいポート（8883）とCA証明書が設定されているか確認する。

入力トピックでメッセージが受信されない ​

説明

  • mqtt-inノードは接続済みだがメッセージが届かない。

考えられる原因

  • パブリッシャーとサブスクライバー間のトピック名不一致
  • QoSレベルの不整合
  • ブローカーのACLルールによるサブスクライブ制限

対処方法

  • パブリッシャーがmqtt-inノードで設定したトピック（test/node_red/in）に正確に送信しているか確認する。
  • Node-REDのデバッグノードを使い、フロー内の各段階でメッセージを検査する。
  • ブローカーの認証およびACL設定を確認する。

フィルターノードがすべてのメッセージをブロックする ​

説明

  • 温度値が変化しても出力トピックにメッセージが現れない。

考えられる原因

  • rbeノードのプロパティパスが誤っている
  • JSONノードがフィルター前にペイロードをパースしていない

対処方法

  • jsonノードがrbeノードの前に配置され、Always Convert to JavaScript Objectに設定されているか確認する。
  • rbeノードのプロパティがmsg.payload.temperature（payload.temperatureではない）に設定されているか確認する。
  • jsonノードの後にdebugノードを追加し、msg.payloadの構造を検査する。

認証に失敗する ​

説明

  • デプロイ直後にノードがdisconnectedとなり、ブローカーのログに認証エラーが記録される。

考えられる原因

  • ブローカー設定でユーザー名またはパスワードが未設定または誤っている
  • トピックに対するACL制限

対処方法

  • ノードをダブルクリックし、ブローカー設定のSecurityタブで正しいユーザー名とパスワードを入力する。
  • EMQXの認証設定を確認する。

さらに詳しく ​

詳細な解説や追加の例については、ブログ記事「Node-REDを使ったMQTTデータ処理」をご覧ください。