メッセージキュー

EMQX 6.0で導入されたメッセージキュー機能は、MQTTのサブスクライブ／パブリッシュパターンを耐久性のあるキューセマンティクスで拡張し、信頼性の高い非同期メッセージ配信を可能にします。RabbitMQのようなエンタープライズグレードのメッセージキューに一般的な機能をネイティブMQTTの機能に追加し、追加のインフラを必要としません。

本ページでは、EMQXのメッセージキュー機能の設計動機、主要概念、内部アーキテクチャ、メッセージフロー、実際の適用シナリオについて包括的に解説します。

メッセージキューとは？

EMQXのメッセージキューは、サブスクライバーの有無に関係なくMQTTメッセージを保存する名前付きの耐久性のあるサーバー側バッファです。各キューは一意のキュー名で識別され、トピックフィルターはどのパブリッシュメッセージをキューに格納するかを定義します（キューの識別子としては機能しません）。設定されたトピックフィルターに一致するメッセージは、キューの保持および配信ポリシーに従って自動的に永続化されます。

従来のMQTTの動作とは異なり、メッセージキューはクライアントがオンラインでない場合でもメッセージを永続化します。クライアントは特別な$queue/<name>または$queue/<name>/<topic_filter>形式にサブスクライブすることでこれらのメッセージを消費できます。

メッセージキューは組み込みの耐久ストレージを使用します。メッセージキューを有効にする前に、EMQXのデータディレクトリがローカルファイルシステムを使用していることを確認してください。組み込み耐久ストレージバックエンドはNFSやSMB/CIFSなどのネットワークファイルシステムをサポートしていません。

リスナーマウントポイントとの非互換性

マウントポイントが設定されたリスナー経由で接続されたクライアントにはメッセージキューは動作しません。EMQXは$queue/プレフィックスのマッチング前にマウントポイントを適用するため、サブスクリプションはマウントされたリテラルトピックへの通常のサブスクリプションとして扱われます。クライアントにエラーは報告されません。

なぜメッセージキューを使うのか？

MQTTは軽量で広く採用されているパブリッシュ／サブスクライブプロトコルですが、デフォルトの動作ではメッセージ配信がサブスクライバーのオンライン状態に強く依存しており、非同期や遅延消費のシナリオでは制約となります。

MQTTの制約

MQTTは共有サブスクリプション（$share/{group}/topic）を通じてキューのような機能を一部サポートしますが、以下の制約があります：

• オンラインのサブスクライバーがいない場合、メッセージは保持されません。
• TTL（有効期限）、キューサイズ制限、オーバーフロー制御の組み込みサポートがありません。
• キーごとに最新の値のみを保持するようなメッセージの重複排除機能がありません。
• キューの明示的なライフサイクル管理がありません。

これらの制約により、以下のようなパターンの実装が困難です：

• デバイスがオンラインになる前にコマンドを送信する。
• 常に接続されていないワーカーにタスクを送る。
• 最新の状態や設定更新のみを保持する。

メッセージキューによるMQTTの拡張

メッセージキューはEMQXのMQTTプロトコルを拡張し、サブスクライバーのオンライン状態に関係なくメッセージを永続化して後続処理を可能にします。主な特徴は：

• クライアントがオフラインでもメッセージを永続化：キューは厳密な順序保証はしませんが、信頼性の高い非同期配信を実現し、軽量なMQTT通信と高度なエンタープライズメッセージングの橋渡しをします。
• 明示的なキュー宣言とプロパティ設定：TTL、サイズ制限、配信戦略などをサポートし、メッセージの保持と配信を細かく制御可能です。
• オプションのラストバリューセマンティクス：同じキーを持つメッセージは前のものを上書きし、最新の状態や設定更新のみを保持できます。

メッセージキューの概念

• キュー名

  メッセージキューを一意に識別する名前です。

  キュー名に使用できる文字は以下のみです：

  • 英数字（A–Z、a–z、0–9）
  • アンダースコア（_）
  • ハイフン（-）
  • ドット（.）

  TIP

  EMQX 6.1.1以降、キューはトピックフィルターではなく名前で指定します。トピックフィルターはキューの設定の一部であり、識別子ではありません。

• トピックフィルター

  devices/+/commandのようなMQTTトピックフィルターで、どのパブリッシュメッセージをキューに書き込むかを決定します。設定されたフィルターに一致するメッセージのみがキューに格納されます。1つのパブリッシュメッセージが複数のキューに一致し、複数のキューに格納されることもあります。

  TIP

  トピックフィルターは名前付きキューの設定メタデータであり、キュー作成後に変更できません。

• キューサブスクリプション

  キューからメッセージを消費するための特別なMQTTサブスクリプションです。クライアントは以下のいずれかの形式でサブスクライブします：

  SUBSCRIBE $queue/<name>
  SUBSCRIBE $queue/<name>/<topic_filter>

  ここで、

  • <name>はキュー名（必須）
  • <topic_filter>は既存キューにサブスクライブする場合は省略可能
  • 自動作成が有効な場合、$queue/<name>/<topic_filter>で指定されたトピックフィルターを使ってキューが存在しなければEMQXが作成します。

  キューサブスクリプションは通常のMQTTサブスクリプションとは独立して動作し、メッセージキューのコンシューマーメカニズムで処理されます。

• ラストバリューセマンティクス

  キュー宣言時にキューキー式を設定して有効化できるオプション機能です。有効化すると、EMQXはキューに入る各メッセージからキューキーを抽出し、同じキーの新しいメッセージが未消費の既存メッセージを上書きします。この動作は状態管理や設定更新のように最新値のみが重要で、古いメッセージを破棄してよい場合に適しています。

  詳細はキューキー式をご参照ください。

• キュー宣言

  耐久性のあるキューを作成し、トピックフィルター、配信戦略、保持制限、キー式などの設定を通じて動作を定義するプロセスです。

• キュー削除

  キューとその保存されたメッセージおよび関連状態をすべて削除する操作です。

• キューのプロパティ

  メッセージ保持時間や配信戦略など、キューの動作を制御するカスタマイズ可能な設定です。

• QoS（サービス品質）

  メッセージキュー内のすべてのメッセージは、パブリッシュやサブスクライブ時のQoSレベルに関わらず、QoS 1（少なくとも1回配信）で配信されます。これにより信頼性の高いメッセージ配信が保証され、キューの配信動作が統一されます。

• メッセージ永続化

  サブスクライバーが接続していなくてもメッセージは保持されます。デフォルトではラストバリューセマンティクスが適用されます。キー式を設定しない通常のキューでは、受信順にメッセージが保存されます。

メッセージキューの動作

EMQXのメッセージキュー機能は疎結合の拡張として実装されており、内部フックを使ってパブリッシュおよびサブスクライブ操作をインターセプトします。これらのフックはレジストリやストレージ層と連携してメッセージを信頼性高く永続化・配信します。

主なコンポーネント

以下の主要コンポーネントが関与します：

• メッセージキューレジストリ：すべてのメッセージキューのライフサイクルを管理し、キューの作成、削除、検索を担当します。
• メッセージキューメッセージDB：キューにパブリッシュされたメッセージを保存し、EMQXの耐久ストレージ上に構築されています。
• メッセージキューステートストレージ：消費進捗やキューメタデータ（TTL、プロパティなど）を永続化します。
• メッセージキューコンシューマー：キューからメッセージを取得し、設定された配信戦略に基づいて接続されたサブスクライバーに配信します。
• メッセージキューサブスクリプションレジストリ：どのチャネル（クライアント）がどのキューにサブスクライブしているかを追跡し、各チャネルのコンテキストにサブスクリプション状態を保存します。
• メッセージキューフック：パブリッシュおよびサブスクライブイベントにフックし、メッセージをキューやコンシューマーにルーティングします。

メッセージキューデータフローダイアグラム

以下の図は主要コンポーネント間のデータフローを示しています：

パブリッシュのワークフロー

1. クライアントがsome/topicのような通常のトピックにメッセージをパブリッシュします。
2. 内部のMQフックがトリガーされ、メッセージを処理します。
3. フックはメッセージキューレジストリで、パブリッシュされたトピックに一致するトピックフィルターを持つキューを検索します。
4. 一致するキューがあれば、メッセージをキューのメッセージDBに書き込みます。

サブスクライブおよび消費のワークフロー

1. クライアントが$queue/<name>または$queue/<name>/<topic_filter>にサブスクライブします。
2. MQフックがサブスクリプションを処理します。
3. フックはキュー名でキューを解決し、クライアントセッションのコンテキスト内にサブスクリプションを初期化し、メッセージキューコンシューマーへの接続を確立します。
4. キューに対応するコンシューマープロセスが存在しなければ、新たにメッセージキューコンシューマーを起動します。
5. コンシューマーはメッセージ消費の進捗を復元し、メッセージDBからデータの取得を開始します。
6. コンシューマーは設定された配信戦略に従い、受信したメッセージをサブスクライバーのクライアントセッションに配信します。
7. サブスクライバーのクライアントセッションは標準MQTTメカニズムを通じてクライアントにメッセージを届けます。

メッセージキューの主要機能

EMQXのメッセージキュー機能は、信頼性が高く疎結合で設定可能なメッセージ配信を実現するコア機能群を提供します。

• メッセージのエンキュー

  キューの設定トピックフィルターに一致するトピックにパブリッシュされたメッセージは自動的にキューに格納されます。

  キューキー式（ラストバリューセマンティクス）を設定している場合、EMQXは各メッセージに対して式を評価します：

  • キーが導出されれば、同じキーの未消費メッセージを置き換えます。
  • ラストバリューキューでキーが評価できなければ、そのメッセージは破棄されます。

• メッセージのデキュー

  サブスクライブしたクライアントは設定された配信戦略に従ってキューからメッセージを受け取ります。メッセージはすべてQoS 1（少なくとも1回配信）で配信され、クライアントがアック（ACK）するとキューから削除されます。

• 配信戦略

  メッセージのサブスクライバー間の配布方法を定義できます：

  • random：ランダムに配布
  • round_robin：利用可能なサブスクライバー間で順番に配布
  • least_inflight：処理中メッセージが最も少ないサブスクライバーを優先

• キュー管理

  キューの作成、更新、削除、クエリなどのライフサイクル操作はREST APIで利用可能です。

ユースケース

メッセージキューは、デバイスやコンシューマーが常にオンラインでない多くのIoTやイベント駆動型アプリケーションで重要な信頼性の高い非同期メッセージングパターンを実現します。

• デバイスコマンドキューイング：クラウドアプリケーションがIoTデバイス向けのコマンドをキューに蓄積し、デバイスがオフラインでもコマンドが失われないようにします。
• バッチ処理：大規模データセットやワークロードを小さなタスクに分割し、ワーカークライアントに並列または遅延処理のために配布します。
• センサーデータ処理：高頻度のセンサーデータを一時的にキューに蓄積し、後でバッチ処理や集約、分析を行います。
• 最新設定の配信：デバイスが常に最新の設定コマンドを取得・処理するようにし、同じ設定項目／キーの古い未処理コマンドはキュー内で上書きまたは無効化されます。

関連機能リファレンス

メッセージキューはMQTTを基盤とし、EMQXの他のメッセージング機能を補完します：

• 共有サブスクリプション：複数のサブスクライバー間でメッセージを分散しますが、クライアントがオンラインでない場合はメッセージを保持しません。
• 保持メッセージ：トピックごとに最後のメッセージを保存しますが、新規サブスクライバーに対して1件のみ配信します。
• MQTT耐久セッション：個々のクライアントのセッション状態（サブスクリプションとQoS 1/2メッセージ）を再接続間で保持します。
• ルールエンジン：SQLライクなルールでキュー内メッセージのフィルタリングや変換、転送を可能にします。

セキュリティ考慮事項

EMQXはキューサブスクリプションの完全なサブスクリプショントピックフィルター（$queue/プレフィックスとキュー名を含む）に対して認可を行います。完全なフィルターに対する書き込み認可ルールを作成してください。キューサブスクリプションは作成前にキューが保存したメッセージも配信します。

キューサブスクリプションは独自のルールが必要

通常のトピックスペース用のルールは対応するキューサブスクリプションをカバーしません：

• t/#のルールは$queue/orders/t/#には適用されません。EMQXはこれらを別のトピックとして扱います。
• #や+で始まる認可トピックフィルターは、$で始まるサブスクリプショントピックフィルターにはマッチしません。デフォルトのacl.confにある{eq, "#"}ルールを含む#を拒否するルールは$queue/orders/#を拒否しません。

#を拒否されたクライアントでも$queue/orders/#にサブスクライブしてキュー内のすべてのメッセージを受信できます。$queue/ネームスペースに対して明示的なルールを追加し、キューのトピックフィルターに対する通常のトピックの認可ルールと同等かそれ以上に厳しくしてください：

%% 自動作成を許可せず、事前作成済みの"orders"キューを1つのコンシューマーに読み取り許可。
{allow, {username, "order_worker"}, subscribe, ["$queue/orders"]}.

%% その他すべてのキューサブスクリプションを拒否（非推奨のプレフィックスも含む）。
{deny, all, subscribe, ["$queue/#", "$q/#"]}.

クライアントがまだ使用している限り、非推奨の$q/プレフィックスはルールに残してください。非推奨プレフィックスも参照してください。

この動作は共有サブスクリプションとは異なります。$share/<group>/t/#の場合、EMQXはプレフィックスを除去してt/#を認可しますが、$queue/<name>/t/#は完全なサブスクリプショントピックフィルターを認可します。

自動作成はクライアント指定のトピックフィルターを許可

キュー自動作成が有効な場合、サブスクライブするクライアントが新規キューのトピックフィルターを決定します。EMQXは$queue/<name>/<topic_filter>の<topic_filter>を使ってキューを作成し、クライアントが直接<topic_filter>にサブスクライブする権限を別途チェックしません。

例えば、$queue/+/#へのサブスクライブを許可されたクライアントは$queue/orders/#にサブスクライブできます。ordersキューが存在しなければ、EMQXは#をトピックフィルターとしてキューを作成します。すると、$で始まらないすべてのトピックのメッセージをキューが保存し、クライアントは直接サブスクライブ権限のないトピックのメッセージを受信する可能性があります。

自動作成はデフォルトでラストバリューキューで有効、通常キューで無効ですが、メッセージキュー機能が有効な間のみ有効です。デフォルトのmq.enable = autoでは、少なくとも1つのキューが存在した後にメッセージキューが有効になるため、サブスクリプションで最初のキューを作成できません。信頼できないクライアントを受け入れる環境では、$queue/ordersのような事前作成済みキューのみアクセス許可し、$queue/#や$q/#にマッチするその他のサブスクリプションはすべて拒否してください。あるいは自動作成を無効にして、ダッシュボードやREST APIからキューを作成してください。ダッシュボードからの自動キュー作成も参照してください。

互換性に関する注意

本節はEMQX 6.1.1で導入された互換性に関する考慮事項をまとめています。

名前付きキュー

EMQX 6.1.1以降、すべてのキューは明示的に名前付きリソースとなりました。キューの識別はトピックフィルターではなく一意の名前に基づきます。

レガシーキュー

以前に作成された名前なしキューには、トピックフィルターに由来する名前が自動的に割り当てられます。

由来名の形式：

/<topic_filter>

この由来名は既存の$q/<topic_filter>サブスクリプションとの後方互換性を維持します。

非推奨プレフィックス

$qプレフィックスはレガシーサブスクリプション向けに引き続きサポートされていますが非推奨です。

新規デプロイメントでは以下を使用してください：

$queue/<name>

共有サブスクリプションの制限

メッセージキューが有効な場合、$queue/プレフィックスはキューサブスクリプション専用に予約され、共有サブスクリプションには使用できません。

次のステップ

メッセージキューの基本を理解したら、実際の利用方法を学びましょう：

• キューの作成と設定：ダッシュボードやREST APIでのキュー宣言、配信戦略や保持ポリシーの設定方法を学べます。
• クイックスタートチュートリアル：MQTTXを使った実践的なパブリッシャー／サブスクライバーシナリオのステップバイステップガイドです。