MQTT Durable Sessions

EMQXにはDurable Sessions機能があり、MQTTセッションおよびメッセージをディスクに永続的に保存することで、高可用性のレプリカを提供し、データの冗長性と一貫性を確保します。セッションの耐久性により、効果的なフェイルオーバーおよびリカバリ機構を実装でき、サービスの継続性と可用性を保証し、システムの信頼性を向上させます。

本ページでは、EMQXにおけるセッション永続化の概念、原理、および使用方法について紹介します。

Important Notice

この機能はEMQX v5.7.0以降で利用可能です。ただし、共有サブスクリプションセッションの永続化はまだサポートされておらず、今後のバージョンでの実装が予定されています。

基本概念

EMQXのDurable Sessions機能を理解する前に、EMQXに関するいくつかの基本概念を押さえておくことが重要です。

セッションと耐久ストレージ

セッション：セッションは、EMQX内でクライアント接続ごとに作成される軽量プロセスです。セッションはMQTT標準によりブローカーに規定された動作を実装しており、初期接続、トピックのサブスクライブおよびサブスクリプション解除、メッセージの配信などを行います。

耐久ストレージ：耐久ストレージはEMQX内部のデータベースです。セッションは自身の状態やトピックに送信されたMQTTメッセージをここに保存できます。耐久ストレージのデータベースエンジンはRocksDBを使用してデータをディスクに保存し、Raftアルゴリズムを用いてクラスター内でデータの一貫したレプリケーションを実現しています。耐久ストレージとDurable Sessionsは混同しないように注意が必要です。

セッション有効期限（Session Expiry Interval）

MQTT標準に従い、クライアントセッションはMQTTブローカー内でクライアント接続と状態の管理を行います。**有効期限（Expiry Interval）**は、クライアント接続終了後にブローカーがセッション状態を保持する期間を制御するセッションのプロパティです。本ドキュメントの文脈で重要な役割を果たします。

有効期限が0に設定されたセッションは、クライアントがEMQXに接続している間のみ存在します。クライアントが切断されると、サブスクリプションや未配信メッセージを含むすべてのセッション情報は即座に破棄されます。一方、有効期限が0より大きいセッションは、クライアントの接続終了後もEMQXに保持されます。クライアントがセッション有効期限内に再接続すると、セッションを再開でき、オフライン時にトピックに送信されたメッセージが配信されます。

• MQTT 5プロトコルを使用するクライアントは、CONNECTまたはDISCONNECTパケットのSession Expiry Intervalプロパティで明示的にセッション有効期限を指定できます。

• MQTT 3.*プロトコルのクライアントの場合、EMQXは以下のルールでセッション有効期限を決定します。Clean Sessionフラグがtrueの場合、有効期限は0に設定されます。それ以外の場合はmqtt.session_expiry_interval設定パラメータの値が使用されます。

EMQXにおけるセッション実装

EMQXは用途に応じて最適化された2種類のクライアントセッション実装を提供しています。

• レギュラーセッション：EMQXノードのメモリ上に状態を保持するセッション。ノード再起動時に状態は失われます。
• Durableセッション：状態および受信メッセージを耐久ストレージにバックアップし、EMQXノード再起動後も再開可能なセッション。

セッション実装の選択は、セッション有効期限とdurable_sessions.enable設定パラメータ（グローバルまたはゾーン単位で設定可能）に依存します。以下の基準で実装が決定されます。

durable_sessions.enable	セッション有効期限 = 0	セッション有効期限 > 0
false	レギュラー	レギュラー
true	レギュラー	Durable

EMQXはメッセージ耐久性管理に独自のアプローチを採用しており、レギュラーセッションとDurableセッションの共存を可能にしつつ、ストレージコストを最小限に抑えています。

セッション実装の比較

クライアントセッションの管理戦略はサービスの安定性と信頼性を確保する上で重要です。本節では2つのセッション実装の特徴を比較し、それぞれの機能や適用シナリオを理解しやすくし、より適切なデプロイ判断を支援します。

レギュラーセッション

この実装はデフォルトであり、EMQX 5.7以前のすべてのリリースで使用されてきました。レギュラーセッションの状態は実行中のEMQXノードのRAMに完全に保持されます。

レギュラーセッションの利点：

• 非常に高いスループットと低いレイテンシ。
• クライアントへの即時メッセージ配信。

欠点：

• セッションデータはセッションをホストするEMQXノードが停止または再起動すると失われる。
• 未配信メッセージはセッションのメモリキューに保存され、EMQXのメモリ使用量が増加する。
• メモリ枯渇を防ぐため、EMQXはメモリキューのサイズに制限を設けており、この制限に達すると新規メッセージは破棄され、未配信メッセージが失われる可能性がある。

Durableセッション

EMQX v5.7.0で導入されたDurableセッションは、セッション状態とDurableセッションにルーティングされたメッセージをディスクに保存します。この機能はデフォルトで無効化されており、durable_sessions.enable設定をtrueにすることで有効化できます。

Durableセッションがトピックフィルターをサブスクライブすると、EMQXはそのフィルターにマッチするトピックを「Durable」とマークします。これにより、これらのトピックからレギュラーセッションへのMQTT PUBLISHメッセージのルーティングに加え、ブローカーはこれらのメッセージをmessagesと呼ばれる耐久ストレージに保存します。

メッセージ配信のプロトコルは、パブリッシャーではなくサブスクライバーのセッションの耐久性に依存する点に注意が必要です。

各Durable MQTTメッセージは、サブスクライブするDurableセッションの数や接続状態に関わらず、各レプリカに一度だけ保存されます。これにより効率的なメッセージファンアウトが可能となり、ディスク書き込みを最小化します。

耐久ストレージは、EMQXクラスター内の複数ノードにセッションメタデータとMQTTメッセージを一貫してレプリケートすることで、高い耐久性と高可用性を提供します。設定可能なレプリケーションファクターにより、各メッセージやセッションのレプリカ数を指定でき、耐久性とパフォーマンスのバランスをユーザーの要件に合わせて調整可能です。

Durableセッションの利点：

• EMQXノードの再起動や停止後もセッションを再開可能。
• MQTTメッセージはメモリキューではなく共有かつレプリケートされた耐久ストレージに保存され、オンライン・オフライン両方のセッションでRAM使用量を削減。
• 未配信メッセージ数に上限がなく、メモリキューのオーバーランによるメッセージ破棄が発生しない。

欠点：

• メッセージをディスクに保存するため、システム全体のスループットは低下する。
• Durableセッションはレギュラーセッションに比べてレイテンシが高い。メッセージの書き込み・読み込みはバッチ処理で行われるため、スループットは向上するが、クライアントがパブリッシュされたメッセージを受信するまでの遅延（エンドツーエンドレイテンシ）が増加する。

Durable Sessionsのクイックスタート

本節では、EMQXおよびMQTTクライアントでDurable Sessions機能を素早く利用する方法を説明し、Durable Sessionsの簡単なワークフローを紹介します。

Note

Durable Sessionsを有効にしていなくても、ステップ2〜4の操作はセッションを保持し、メッセージはクライアントキューに保存されます。違いは、セッションが永続的に保存されるかどうか、およびノード再起動後にセッションが復元可能かどうかです（ステップ5）。

1. EMQXでDurable Sessions機能を有効化します。

   デフォルトではDurable Sessionsは無効です。etc/emqx.confファイルを編集し、以下の設定を追加して機能を有効化してください。

   durable_sessions {
     enable = true
   }

   設定を反映するためにEMQXを再起動します。

2. MQTTクライアントの接続パラメータを調整し、セッション耐久性を有効にします。

   例としてMQTTX CLIを使用します。MQTT 5.0プロトコルがデフォルトで、--no-cleanオプションを付けてClean Start = falseを設定し、クライアントIDをemqx_cに指定してEMQXに接続し、t/1トピックをサブスクライブします。

   mqttx sub -t t/1 -i emqx_c --no-clean

3. クライアントを切断し、セッションを保持します。

   ステップ2のクライアントを切断します。EMQXダッシュボードのモニタリング -> クライアントページを開くと、クライアントの状態がDisconnectedのまま表示され、セッションが保持されていることがわかります。

   

4. クライアントにメッセージを送信し、メッセージがクライアントキューに保存されます。

   再度MQTTX CLIを使用し、benchコマンドで1クライアントからt/1トピックに繰り返しメッセージをパブリッシュします。

   mqttx bench pub -t t/1 -c 1

   MQTTプロトコルに従い、emqx_cクライアントがオフラインでも、サブスクライブしている t/1トピックのメッセージはクライアントキューに保存され、再接続時に配信されます。

5. EMQXノードを再起動し、セッションとメッセージが耐久ストレージから復元されます。

   EMQXノードを再起動します。クライアントの接続操作を行わずにEMQXダッシュボードのモニタリング -> クライアントページを開くと、クライアントがDisconnected状態で表示され、セッションが復元されていることが確認できます。

   同じクライアントID emqx_c と --no-clean オプションを指定してClean Start = falseでEMQXに接続を試みます。

   mqttx sub -t t/1 -i emqx_c --no-clean

   オフライン期間中に受信したメッセージがクライアントに配信されます。

   ...
   [2024-5-22] [16:14:14] › …  Connecting...
   [2024-5-22] [16:14:14] › ✔  Connected
   [2024-5-22] [16:14:14] › …  Subscribing to t/1...
   [2024-5-22] [16:14:14] › ✔  Subscribed to t/1
   [2024-5-22] [16:14:14] › payload: Hello From MQTTX CLI
   ...

   Note

   • セッションを復元するには、同じクライアントID emqx_c を使用し、--no-cleanオプションでClean Startをfalseに設定する必要があります。この2つの条件を満たす必要があります。
   • 以前のサブスクリプション情報はすでにセッションに保存されているため、再接続時に t/1トピックを再サブスクライブしなくてもメッセージは配信されます。

耐久ストレージのアーキテクチャ

EMQXの組み込み耐久機能を支えるデータベースエンジンは、ストレージ、シャード、ジェネレーション、ストリームという階層構造でデータを管理しています。

ストレージ

ストレージは、MQTTメッセージやMQTTセッションなど、特定の種類のすべてのデータをカプセル化します。

シャード

メッセージはパブリッシャーのクライアントIDに基づいてクライアントごとに分割され、シャードに格納されます。シャード数はEMQXの初回起動時にn_shards設定パラメータで決定されます。シャードはレプリケーションの単位でもあり、各シャードはdurable_storage.messages.replication_factorで指定された回数だけ異なるノードに一貫してレプリケートされ、各レプリカに同一のメッセージセットを保持します。

ジェネレーション

シャード内のメッセージは特定の時間枠に対応するジェネレーションに分割されます。新しいメッセージは現在のジェネレーションに書き込まれ、過去のジェネレーションは読み取り専用です。EMQXは古いジェネレーションを丸ごと削除することで古いMQTTメッセージをクリーンアップします。古いMQTTメッセージの保持期間はdurable_sessions.message_retention_periodパラメータで決まります。

ジェネレーションはストレージレイアウト仕様に応じて異なる方法でデータを整理できます。現在はワイルドカードおよび単一トピックのサブスクリプションに最適化された1つのレイアウトのみがサポートされています。将来的には異なるワークロードに最適化されたレイアウトが追加される予定です。

新しいジェネレーションのストレージレイアウトはdurable_storage.messages.layoutパラメータで設定され、各レイアウトエンジンは独自の設定パラメータを持ちます。

ストリーム

各シャードおよびジェネレーション内のメッセージはストリームに分割されます。ストリームはEMQXにおけるメッセージのシリアライズ単位です。ストリームは複数のトピックのメッセージを含むことができます。異なるストレージレイアウトはトピックをストリームにマッピングするための異なる戦略を用います。

Durableセッションはストリームからバッチ単位でメッセージを取得し、バッチサイズはdurable_sessions.batch_sizeパラメータで調整可能です。

クラスター全体の耐久ストレージ

EMQXクラスター内の各ノードには一意のサイトIDが割り当てられており、これはErlangノード名（emqx@...）とは独立した安定した識別子です。サイトIDは永続的であり、ノードの初回起動時にランダムに生成されます。この安定性により、ノード名の変更や再設定があってもデータの整合性が保たれます。

管理者はemqx_ctl ds info CLIコマンドを使用してクラスター全体の耐久ストレージの状態を管理・監視できます。

Durable Sessionsのハードウェア要件

セッション耐久性を有効にすると、EMQXはDurableセッションのメタデータおよびDurableセッションに送信されたMQTTメッセージをディスクに保存します。そのため、十分なストレージ容量を持つサーバーにEMQXをデプロイする必要があります。最高のスループットを得るためには、ソリッドステートドライブ（SSD）の使用を推奨します。

ディスク容量の目安は以下の通りです：

• メッセージストレージ：各レプリカに保存されるメッセージの容量は、受信メッセージのレートとdurable_sessions.message_retention_periodパラメータで指定された保持期間の積に比例します。このパラメータはメッセージの保持期間を決定し、必要なストレージ容量に影響します。
• セッションメタデータストレージ：セッションメタデータの容量は、セッション数とそれらがサブスクライブするストリーム数の積に比例します。
• ストリームの計算：ストリーム数はシャード数に比例し、トピック数にも（非線形に）依存します。EMQXは構造が類似したトピックを自動的に同じストリームにまとめるため、トピック数の増加に伴うストリーム数の増加を抑制し、セッションごとのメタデータ量を最小化します。

次のステップ

Durable Sessions機能の設定および管理方法、EMQXクラスターでのDurable Sessionsの初期設定や変更方法については、以下のページをご参照ください。

• Durable Sessionsの設定と管理
• データレプリケーションの管理