Data Integrationによるクライアント接続・切断イベントトピックメッセージの取得
クライアント接続および切断イベントトピックは、EMQXがクライアントの接続状態の変化を通知するために使用するシステムトピックで、$events/client_connectedや$events/client_disconnectedなどがあります。通常のクライアントはこれらのトピックを直接サブスクライブできませんが、ルールエンジンを通じてキャプチャし、データベースへの書き込みや他のトピックへの転送、リアルタイム監視・分析などに活用できます。
EMQXはMQTTブローカーとして機能し、ストレージサービスではないため、セキュリティおよびプライバシーの観点から、デフォルトでは過去のクライアント状態の記録やデバッグレベルのイベントログを保持しません。したがって、デプロイメントにおいてクライアント接続・切断イベントトピック用のデータ統合ルールを設定し、デバイスの接続状態をリアルタイムで追跡し、トラブルシューティングや監査に役立つ重要な活動記録を保存することを推奨します。
本ページでは、クライアント接続・切断イベントトピックのユースケースを紹介し、ルールエンジンを用いたメッセージのキャプチャ方法を解説します。また、MQTTX Desktopを使ったイベントトピックメッセージの受信シミュレーションも示します。
主なユースケース
クライアント接続および切断イベントトピックは、以下のような複数のシナリオで利用可能です。
- デバイス接続履歴の記録：イベントを外部データベースにログとして記録し、デバイスの接続履歴や動作パターンを追跡、後の分析やレポート作成に活用。
- デバイスのオンライン状態および活動の監視：デバイスの接続状況やクライアントの活動をリアルタイムに把握し、業務システムの動的な調整やポリシー決定を支援。
- 切断トラブルシューティング：切断イベントの原因や発生タイミングを分析し、技術サポートチームが予期せぬ切断を迅速に特定・解決。
- セキュリティ監査およびアクセス行動の追跡：イベントをログプラットフォームに保存し、アクセス追跡やセキュリティ調査、コンプライアンス監査に利用。
トリガー条件と主なフィールド
クライアント接続イベント（"$events/client_connected"）
- トリガー：クライアントが正常に接続したときに発生。
- 主なフィールド：
| フィールド名 | 説明 |
|---|---|
| clientid | 接続したクライアントのID |
| username | 接続に使用されたユーザー名 |
| peername | クライアントのIPアドレスとポート |
| keepalive | MQTTのキープアライブ間隔（クライアントからブローカーへのハートビート頻度） |
| clean_start | MQTTのclean_start（クライアントがクリーンセッションを開始するかどうか） |
| expiry_interval | MQTTセッションの有効期限（切断後にセッションが保持される期間） |
| connected_at | 接続が確立したタイムスタンプ（単位：ミリ秒） |
クライアント切断イベント（"$events/client_disconnected"）
- トリガー：クライアントが切断したときに発生。
- 主なフィールド：
| フィールド名 | 説明 |
|---|---|
| reason | 切断の理由 |
| clientid | 切断したクライアントのID |
| username | 接続に使用されたユーザー名 |
| peername | クライアントのIPアドレスとポート |
| disconnected_at | クライアントが切断したタイムスタンプ（単位：ミリ秒） |
TIP
イベントトピックの全フィールド一覧については、公式ドキュメントのイベントトピック一覧を参照してください。
切断シナリオの分析
クライアント切断イベントトピックのreasonフィールドは切断の具体的な原因を示し、接続問題の理解やクライアント接続ロジックの最適化に役立ちます。以下に切断理由の詳細を示します。
| 理由コード | 説明 |
|---|---|
| normal | クライアントが能動的に切断した場合。通常のMQTTプロトコルに従いセッションを終了する標準的なシナリオです。 |
| kicked | サーバー側から切断された場合。通常は管理者がAPIやコンソールからセッションを強制終了した場合に発生します。 |
| keepalive_timeout | クライアントが指定されたキープアライブ間隔内にハートビート（Pingreq）をEMQXに送信できなかった場合に発生。主な原因は以下の通りです。 1. ネットワークジッター：クライアントとEMQX間のネットワークが不安定でハートビートメッセージが遅延・欠落。 2. クライアントのハートビート不具合：クライアントのハートビート機構が正常に動作せず、必要な間隔でMQTTハートビートを送信しない。 詳細はキープアライブのドキュメントを参照してください。 |
| not_authorized | クライアントの認証失敗または権限不足による切断。主な原因は以下の通りです。 1. 認証失敗：クライアントが無効なMQTT認証情報を提供。 2. 権限不足： acl_nomatch = disconnect（デフォルト）設定時に、適切なアクセス権限なしにパブリッシュやサブスクライブを試行。 |
| tcp_closed | リモート側がTCP接続を閉じた場合。クライアントクラッシュやネットワーク障害により、MQTTのDISCONNECTパケットを経ずに接続が切断されることが多いです。例として、クライアントアプリの強制終了、ISPによるネットワーク遮断、デバイスのネットワーク切断などがあります。 |
| internal_error | パケットの不正フォーマットやその他不明なエラーによる切断。解決にはパケット内容の確認、サーバーログの調査、EMQX技術サポートへの問い合わせが必要です。 |
| discarded | clean_start = trueの状態で同一ClientIDを持つ2つのクライアントが接続した場合に発生。EMQXは既存のセッションを破棄し、新しいセッションを確立し、古いクライアントを切断します。clean_start = trueはクライアントが毎回新しいセッションを開始したいことを示します。同一ClientIDの新規接続があると、EMQXは古いサブスクリプションやメッセージキューを削除し、古いセッションを新しい接続に置き換えます。 |
| takenover | clean_start = falseの状態で同一ClientIDを持つ2つのクライアントが接続した場合に発生。EMQXは新しいクライアントに既存セッションを引き継がせ、古いクライアントを切断します。clean_start = falseはクライアントが既存セッションを再利用したいことを示します。同一ClientIDの新規接続があると、EMQXは古いセッションを新しいクライアントに移行し、前のクライアントを切断します。 |
TIP
以下はClientID競合による切断の代表的な2つのシナリオです。
ClientID競合によるセッション奪取の繰り返し：複数の異なるクライアントが同一ClientIDを使用し、自動再接続が有効な場合、新しい接続が古いセッションを繰り返し切断し、再接続ループが発生します。この場合、クライアントのIPやポートが頻繁に変わることが多いです。
推奨：各クライアントにユニークなClientIDを割り当ててください。
クライアントの再接続が繰り返される場合：クライアントセッションが残っているにもかかわらず、クライアント側がそれをアクティブと認識せず、新たにMQTT CONNECTパケットを繰り返し送信し、既存セッションを上書きすることがあります。この場合、IPは同じままでポートが変わることが多く、一部のIoTネットワークモジュールではIPとポートの両方が変わることもあります。
推奨：クライアントの接続ロジックを見直し、最適化してください。
イベントトピックメッセージをキャプチャするためのデータ統合設定
実際には、イベントトピックの処理は主に以下の2つの方法で行われます。
- メッセージの再パブリッシュ：イベントトピックメッセージを他のMQTTトピックに再パブリッシュします。この方法は軽量でリアルタイム性が高く、ネイティブなMQTTエコシステムと完全に互換性があるため、システム内でイベントを迅速に消費・処理するのに適しています。
- 外部サービスへの転送：イベントトピックメッセージをデータベース、メッセージキュー、HTTPサービスなどの外部システムに送信します。この方法により多様な外部システムとの連携が可能となり、リアルタイム応答や永続的な処理をサポートします。
本ページでは、メッセージの再パブリッシュとHTTPサービスへの転送を例示します。その他の方法（異なるデータベースや外部システムへの転送など）については、公式ドキュメントのEMQX Cloud データ統合を参照してください。
メッセージの再パブリッシュ
クライアント接続・切断イベントトピックメッセージを他のトピックに再パブリッシュする方法を示します。
ルールとアクションの作成
- データ統合ページのデータ転送カテゴリでRepublishをクリックします。既存のコネクターがある場合は新規コネクターをクリックし、データ転送カテゴリの中からRepublishを選択します。
- SQLエディターでルールSQLを定義します。クライアント切断イベントを追跡するには、以下のSQL例を使用できます。
SELECT
clientid,
username,
reason,
disconnected_at
FROM
"$events/client_disconnected"- 次へをクリックしてアクションを追加します。
- アクション作成画面で以下を設定します。
- コネクター：デフォルトのRepublishを使用。
- トピック：ターゲットトピックを
client_disconnectedに設定。 - ペイロード、QoS、Retain：デフォルト値を使用。
- 確認をクリックして設定を完了します。
ルールとアクションのテスト
MQTTXを使ってクライアント接続やメッセージ送信をシミュレートすることを推奨しますが、任意のMQTTクライアントでも構いません。
- MQTTXでデプロイメントに2つの接続を作成し、ClientIDをそれぞれ
connとsubに設定します。 - クライアント
subで再パブリッシュされたトピックclient_disconnectedをサブスクライブします。 - クライアント
connを切断します。クライアントsubは以下の形式のイベントメッセージを受信します。
{
"clientid": "conn",
"username": "u_emqx",
"reason": "normal",
"disconnected_at": 1645003578536
}イベントトピックメッセージをHTTPサービスに転送
クライアント接続・切断イベントトピックメッセージをHTTPサービスに転送する方法を示します。開始前に、ターゲットコネクターへ内部IPでアクセスするためのVPCピアリング接続か、パブリックIPでアクセスするためのNATゲートウェイを用意してください。
HTTPサーバーコネクターの作成
デプロイメントメニューでデータ統合を選択し、Webサービスカテゴリの中からWebhookを選びます。既存のコネクターがある場合は新規コネクターをクリックし、Webサービスカテゴリの中からWebhookを選択します。
コネクター作成画面でBase URLを入力し、その他の設定は必要に応じて調整します。
URLは転送先のHTTPサービスを指し、コネクターはルールで定義されたペイロードをPOSTリクエストとしてこのURLに送信します。
テストをクリックして、EMQXが指定URLに正常にアクセスできるか確認します。
新規作成をクリックしてコネクター作成を完了します。
ルールとアクションの作成
- 新規ルールをクリックし、ルール作成画面に入ります。
- SQLエディターでルールSQLを定義します。クライアント切断イベントを追跡するには、以下のSQL例を使用できます。
SELECT
clientid,
username,
reason,
disconnected_at
FROM
"$events/client_disconnected"- 次へをクリックしてアクション作成を開始します。
- コネクターのドロップダウンから先ほど作成したHTTPサーバーコネクターを選択し、その他の設定はデフォルトのままにします。
- 確認ボタンをクリックしてルール作成を完了します。
TIP
HTTPサービスへのデータ転送の全手順と設定については、公式ドキュメントのHTTPサーバーへのMQTTデータ取り込みを参照してください。
ルールとアクションのテスト
MQTTXを使ってクライアント接続やメッセージ送信をシミュレートすることを推奨しますが、任意のMQTTクライアントでも構いません。
- MQTTXでデプロイメントに接続し、ClientIDを
connに設定します。 - クライアント
connを切断します。HTTPサービスは以下の形式のPOSTリクエストを受信します。
{
"clientid": "conn",
"username": "u_emqx",
"reason": "normal",
"disconnected_at": 1645003578536
}