Data Integration を使ったメッセージパブリッシュおよびドロップイベントトピックメッセージのキャプチャ

メッセージパブリッシュおよびドロップイベントトピックは、EMQX が提供するシステムイベントトピックであり、ブローカー内のメッセージのライフサイクル（パブリッシュ、配信、アック、ドロップイベント）をリアルタイムに監視するためのものです。これらのイベントは、メッセージのフロー解析、メッセージロスの診断、メッセージスループットの測定に不可欠です。通常のクライアントはこれらのシステムトピックを直接サブスクライブできませんが、ルールエンジンを使ってキャプチャし、データベースへの保存や他のトピックへの転送、リアルタイム監視や分析に活用できます。

EMQX ブローカーは MQTT ブローカーとして機能し、ストレージサービスではないため、セキュリティやプライバシーの観点から、メッセージ送信やドロップの履歴をデフォルトで保存しません。したがって、メッセージパブリッシュおよびドロップイベントトピックに基づく Data Integration ルールを設定することを推奨します。これにより、メッセージのライフサイクル情報、メッセージの信頼性指標、システムの配信品質をリアルタイムに収集・分析・永続化でき、パフォーマンス最適化、トラブルシューティング、監査分析に役立ちます。

本ページでは、メッセージパブリッシュおよびドロップイベントトピックのユースケースを説明し、ルールエンジンを通じてイベントメッセージをキャプチャし処理する方法を示し、MQTTX Desktop を使ったイベントメッセージをトリガーするクライアントのシミュレーション方法を紹介します。

ユースケース

メッセージパブリッシュおよびドロップイベントトピックは、以下のような幅広い運用およびビジネスシナリオで活用できます。

メッセージトレーシングおよび配信経路監視：クライアントによるパブリッシュから正常配信までのメッセージライフサイクル全体を追跡し、メッセージの蓄積やレイテンシ、配信失敗を特定します。
ビジネスメトリクスおよびスループット分析：クライアントやトピックごとのパブリッシュ数や配信数をカウントし、アプリケーション負荷やユーザー行動パターンを把握します。
異常検知およびアラート：サブスクライバー不在や配信失敗、キューオーバーフローによるメッセージドロップを検知し、イベントトピック経由で詳細なドロップ情報を取得します。
セキュリティ監査およびアクセスログ：どのクライアントがどのメッセージをパブリッシュし、正常に配信・消費されたかを記録し、監査やコンプライアンス対応を支援します。

トリガー条件と主なフィールド

ルーティング中にメッセージがドロップされた場合 (`$events/message_dropped`)

トリガー条件：メッセージにマッチするサブスクライバーが存在しない場合に発生します。
主なフィールド：

フィールド名	説明
clientid	メッセージパブリッシャーのクライアントID
username	メッセージパブリッシャーのユーザー名
topic	メッセージのトピック
reason	メッセージがドロップされた理由
qos	メッセージの QoS レベル
timestamp	ドロップイベントのタイムスタンプ（ミリ秒）
publish_received_at	PUBLISH パケットがブローカーに到達したタイムスタンプ（ミリ秒）

主なドロップ理由：

理由コード	説明
no_subscribers	トピックにサブスクライバーが存在しない
receive_maximum_exceeded	QoS 2 の awaiting_rel キューが満杯
packet_identifier_inuse	QoS 2 用のパケットIDがまだ使用中で解放されていない

典型的なユースケース：サブスクライバー不在によるメッセージドロップの監視。

配信中にメッセージがドロップされた場合 (`$events/delivery_dropped`)

トリガー条件：サブスクライバーのメッセージキューが満杯の場合に発生します。
主なフィールド：

フィールド名	説明
from_clientid	メッセージパブリッシャーのクライアントID
from_username	メッセージパブリッシャーのユーザー名
topic	メッセージのトピック
reason	メッセージがドロップされた理由
qos	メッセージの QoS レベル
timestamp	ドロップイベントのタイムスタンプ（ミリ秒）
publish_received_at	PUBLISH パケットがブローカーに到達したタイムスタンプ（ミリ秒）

主なドロップ理由：

理由コード	説明
no_local	クライアントが自身のパブリッシュしたメッセージを受信できない（MQTT 5 の No Local = 1）
expired	メッセージまたはセッションの有効期限切れ
queue_full	サブスクライバーのメッセージキューが満杯（消費遅延や長期間オフラインのため）
qos0_msg	QoS 0 メッセージがセッションメッセージキューに保存されない設定のため破棄された。詳細はセッション設定参照。

典型的なユースケース：
- サブスクライバーの消費能力やオフライン期間の監視
- メッセージバックログや配信失敗原因の分析
- $events/message_dropped と組み合わせてアップリンク・ダウンリンク双方のメッセージドロップ挙動を完全に追跡

メッセージ配信完了イベント (`$events/message_delivered`)

トリガー条件：EMQX がメッセージをサブスクライバーに正常に送信した時点で発生（クライアントのアックではなく、ブローカー側の配信完了を示す）。
主なフィールド：

フィールド名	説明
clientid	配信先クライアントID
from_clientid	メッセージパブリッシャーのクライアントID
username	配信先クライアントのユーザー名
topic	メッセージのトピック
qos	メッセージの QoS レベル
timestamp	イベントのタイムスタンプ（ミリ秒）
publish_received_at	PUBLISH パケットがブローカーに到達したタイムスタンプ（ミリ秒）

典型的なユースケース：
- メッセージ配信成功率の測定
- メッセージ配信レイテンシの監視
- $events/message_acked と組み合わせてアックレイテンシ分析（QoS 1 / QoS 2）

メッセージアックイベント (`$events/message_acked`)

トリガー条件：クライアントが PUBACK または PUBCOMP を送信して受信をアックした時（QoS 1 および QoS 2 のみ）。
主なフィールド：

フィールド名	説明
id	MQTT メッセージID（レイテンシ追跡に有用）
clientid	配信先クライアントID
from_clientid	パブリッシャークライアントID
username	配信先クライアントのユーザー名
topic	メッセージのトピック
qos	メッセージの QoS レベル
timestamp	イベントのタイムスタンプ（ミリ秒）
publish_received_at	PUBLISH パケットがブローカーに到達したタイムスタンプ（ミリ秒）

典型的なユースケース：
1. メッセージ配信とアック間のレイテンシ計測
2. QoS 1 / QoS 2 メッセージのアック率分析
3. ミッションクリティカルなワークフローにおける信頼性配信の保証

TIP

イベントトピックの全フィールド一覧は、MQTT イベントを参照してください。

Data Integration でドロップイベントトピックメッセージをキャプチャする設定

実際のデプロイでは、イベントトピックの処理は主に以下の2通りです。

メッセージの再パブリッシュ：イベントメッセージを別の MQTT トピックに再パブリッシュします。
軽量かつリアルタイムで、ネイティブな MQTT ワークフローと親和性があります。
外部サービスへの転送：イベントメッセージをデータベースやメッセージキュー、HTTP サービスなど外部システムに送信します。
永続的な分析や下流システムとの統合が可能です。

ここではメッセージの再パブリッシュ方法のみを示します。外部サービスへの転送方法は、メッセージ配信完了イベントメッセージを HTTP サービスに転送するを参照してください。

その他のデータベースや外部サービスへの転送方法については、公式ドキュメントのEMQX Cloud Data Integrationを参照してください。

メッセージ再パブリッシュ用のルールとアクションを作成する

ここでは、メッセージドロップイベントトピックメッセージを別トピックに再パブリッシュする例を示します。

ルールとアクションの作成手順

Data Integration の Data Forward から Republish を選択します。既にコネクターが存在する場合は、Create Connector をクリックして Republish を選択してください。
SQL エディターでルールの SQL を定義します。例えば、メッセージドロップイベントを診断するには以下のように記述します。
sql
```
SELECT
    clientid,
    reason,
    topic,
    qos,
    timestamp
FROM
    "$events/message_dropped"
```
Next をクリックしてアクションを追加します。
アクションを設定します。
- Connector：デフォルトの Republish を使用します。
- Topic：ターゲットトピックを message_dropped に設定します。
- Payload、QoS、Retain：デフォルト値のままにします。
Confirm をクリックして設定を完了します。

ルールとアクションのテスト

MQTTXや任意の MQTT クライアントを使用できます。

ClientID が pub と sub の2つの MQTTX 接続を作成します。
sub クライアントが message_dropped をサブスクライブします。
pub がトピック test にメッセージをパブリッシュします。sub クライアントは以下のようなメッセージを受信します。

json

{
    "topic": "test",
    "timestamp": 1761309999897,
    "reason": "no_subscribers",
    "qos": 1,
    "clientid": "pub"
}

このペイロードは、メッセージが EMQX に到達した時点でトピックにサブスクライバーが存在しなかったためドロップされたことを示しています。

メッセージ配信完了イベントメッセージを HTTP サービスに転送する

ここでは、メッセージ配信完了イベントメッセージを HTTP サービスに転送する方法を示します。開始前に、プライベートIPでコネクターにアクセスするためのVPC ピアリング接続を作成するか、パブリックIPでアクセスするためにNAT ゲートウェイを有効にしてください。

HTTP サーバーコネクターを作成する

Data Integration の Web Services から HTTP Server を選択します。既にコネクターがある場合は、Create Connector を選択し、HTTP Server を選びます。
転送先 HTTP エンドポイントの URL を入力し、必要に応じてその他の設定を調整します。
URL は、転送先の HTTP サービスを指し、コネクターはルールで定義されたペイロードを POST リクエストで送信します。
Test をクリックして接続を検証します。
New をクリックしてコネクター作成を完了します。

ルールとアクションを作成する

Create Rule をクリックします。

SQL エディターで以下のように SQL を定義します。

sql

SELECT
    from_clientid,
    clientid,
    username,
    payload,
    topic,
    qos
FROM
    "$events/message_delivered"

Next をクリックしてアクションを作成します。
先ほど作成した HTTP Server コネクターを選択し、その他の設定はデフォルトのままにします。
Confirm をクリックします。

TIP

HTTP サービスへのデータ転送の全手順と設定については、公式ドキュメントのHTTP Server への MQTT データ取り込みを参照してください。

ルールとアクションのテスト

ClientID が pub と sub の2つの MQTTX 接続を作成します。
sub がトピック test をサブスクライブします。
pub が test にメッセージをパブリッシュします。
HTTP サービスは以下のようなペイロードを含む POST リクエストを受信します。

json

{
    "username": "test",
    "topic": "test",
    "qos": 0,
    "payload": "Hello Event",
    "from_clientid": "pub",
    "clientid": "sub"
}

ベストプラクティスの推奨事項

高頻度イベント（例：message_delivered）の場合は、レートリミットを有効にするか、ルール SQL の WHEN 句でイベントをフィルタリングするか、Kafka などの外部システムに転送してバッファリングすることを推奨します。
イベントトピックやビジネスモジュールごとに別々のルールとコネクターを使用し、保守性と分析の分離性を向上させてください。
クライアントのオンライン／オフラインイベントトピックと組み合わせて、デバイスの接続状態およびメッセージ挙動のライフサイクル全体を監視してください。

専用フレックス

VPC ピアリング接続

PrivateLink の設定

拡張認証

拡張認可

アラート

アラートトラブルシューティングガイド

ルール

LLM ベースの MQTT データ処理

サーバレスデプロイメント API

Data Integration を使ったメッセージパブリッシュおよびドロップイベントトピックメッセージのキャプチャ ​

ユースケース ​

トリガー条件と主なフィールド ​

ルーティング中にメッセージがドロップされた場合 ($events/message_dropped) ​

配信中にメッセージがドロップされた場合 ($events/delivery_dropped) ​

メッセージ配信完了イベント ($events/message_delivered) ​

メッセージアックイベント ($events/message_acked) ​

Data Integration でドロップイベントトピックメッセージをキャプチャする設定 ​

メッセージ再パブリッシュ用のルールとアクションを作成する ​

ルールとアクションの作成手順 ​

ルールとアクションのテスト ​

メッセージ配信完了イベントメッセージを HTTP サービスに転送する ​

HTTP サーバーコネクターを作成する ​

ルールとアクションを作成する ​

ルールとアクションのテスト ​

ベストプラクティスの推奨事項 ​

Data Integration を使ったメッセージパブリッシュおよびドロップイベントトピックメッセージのキャプチャ

ユースケース

トリガー条件と主なフィールド

ルーティング中にメッセージがドロップされた場合 (`$events/message_dropped`)

配信中にメッセージがドロップされた場合 (`$events/delivery_dropped`)

メッセージ配信完了イベント (`$events/message_delivered`)

メッセージアックイベント (`$events/message_acked`)

Data Integration でドロップイベントトピックメッセージをキャプチャする設定

メッセージ再パブリッシュ用のルールとアクションを作成する

ルールとアクションの作成手順

ルールとアクションのテスト

メッセージ配信完了イベントメッセージを HTTP サービスに転送する

HTTP サーバーコネクターを作成する

ルールとアクションを作成する

ルールとアクションのテスト

ベストプラクティスの推奨事項