EMQX クラスタリングの設計
MQTT はステートフルなプロトコルであり、ブローカーは各 MQTT セッションの状態情報（サブスクライブされたトピックや未完了のメッセージ送信など）を保持する必要があります。MQTT ブローカーのクラスタリングにおける主な課題の一つは、これらの状態をすべてのクラスタノード間で効率的かつ信頼性高く同期・複製することです。
EMQX は高いスケーラビリティとフォールトトレランスを備えた MQTT ブローカーであり、複数のノードでクラスタモードで動作可能です。EMQX のクラスタリングは、IoT メッセージングシステムのスケーラビリティ、可用性、信頼性、管理性を向上させ、大規模またはミッションクリティカルな用途に推奨されるアプローチです。本ページでは、MQTT ブローカーのクラスタリングの必要性と EMQX がどのようにこれを実現し、単一クラスタ内で数百万のユニークなワイルドカードサブスクライバーをサポートできるかを解説します。
EMQX クラスタの作成と運用の詳細な手順については、EMQX Cluster をご参照ください。
クラスタリングの重要なポイント
クラスタ設計において考慮すべき重要なポイントがいくつかあります。これらはクラスタの成功を左右する最も重要な要素です。簡単にまとめると以下の通りです。
集中管理：クラスタ内のすべてのノードを単一の管理コンソールから監視・制御できるようにすること。
データの一貫性：クラスタ内のすべてのノードがルーティング情報を一貫して保持できるように、データを全ノードに複製すること。
スケーラビリティの容易さ：クラスタ管理の複雑さを減らすため、新しいノードの追加が複雑でないこと。クラスタは新規ノードを自動検出し、クラスタに組み込めること。
クラスタのリバランス：運用オーバーヘッドを最小限に抑えつつ、各ノードの負荷の偏りを検知し、負荷の少ないノードへワークロードを再割り当てできること。これにより、1台以上のノードが故障してもクラスタが継続稼働可能となる。
大規模クラスタサイズ：システムの増大する要求に応じてノードを追加し、水平スケールできること。
自動フェイルオーバー：ノード障害時にクラスタが自動で検知し、残りのノードにワークロードを再割り当てできること。
ネットワークパーティション耐性：ネットワークパーティションが発生してもクラスタが継続稼働できること。
EMQX はこれらの目標を最も効率的に達成するためにさまざまな手法を用いています。以下のセクションでクラスタリングの主要な側面を詳述します。
データ複製チャネル
メタデータおよびメッセージの複製を実現するために、Erlang 分散プロトコルとカスタム分散プロトコルがブローカー間のリモートプロシージャコールに利用されています。EMQX クラスタには2つのデータ複製チャネルがあります。
メタデータ複製：どのノードがどの（ワイルドカード）トピックをサブスクライブしているかなどのルーティング情報。これは「Erlang 分散」プロトコルにより実現され、各ノードはクライアント兼サーバーとして機能します。このプロトコルのデフォルトリスニングポートは 4370 です。
メッセージ配信：ノード間でメッセージを転送する際に使用されます。メッセージ配信チャネルはコネクションプールを用い、各ノードはデフォルトでポート番号 5370（Docker コンテナ環境では 5369）をリッスンします。Erlang 分散プロトコルとは異なり、こちらは単一コネクションではなく複数コネクションを利用します。
下図は、2つのデータ複製チャネルとパブリッシュ・サブスクライブのフローを示しています。点線はメタデータ複製を、実線矢印はメッセージ配信チャネルを表しています。
組み込みデータベース
EMQX は内部データを2種類の組み込みデータベース管理システムのいずれかに格納します。
Mria：軽量なインメモリデータベースで、ルーティングテーブルやランタイム設定など読み込みが多いワークロードに使用されます。CAP 定理においては 可用性 を重視した設計です。
Durable Storage (DS)：ディスクベースのストリーミングデータベースで、耐久セッションやメッセージキューなど、書き込みが多く大量のデータに対応します。こちらは 一貫性 を重視した設計です。
これら2つのデータベース管理システムは性質が大きく異なり、ネットワークパーティション時の保証も異なります。本ドキュメントでは主に Mria と EMQX の動作に焦点を当て、耐久性機能の詳細は扱いません。耐久ストレージに関する情報は Durable Storage をご参照ください。
Mria テーブルはさらに以下の2種類に分類されます。
Regular：テーブルの内容がグローバルに一様であるもの。Mria テーブルの大半はこちら。
Merge：各レコードが特定の EMQX ノードに帰属し、そのノードのみが書き込み可能。他ノードからは読み取り専用として見える特殊なテーブル。
ノードの役割：Core と Replicant
Mria は core と replicant の2種類のノード役割を持つ混合ネットワークトポロジーを採用しています。
EMQX クラスタには少なくとも1つの Core ノードが必要で、Replicant ノードは任意の数を配置可能です。
Core ノードは Mria の中核であり、regular テーブルの更新を調整します。これらのテーブルの更新は Core ノード間で同期的に複製されます。調整コストがかかるため、Core ノードは冗長性要件を満たす最小限の数（通常は3台）に抑え、残りは Replicant 役割を割り当てることが推奨されます。
Replicant ノードはトランザクション処理に直接関与せず、Core ノードに接続してデータ更新を受動的に複製します。書き込み操作は許可されず、書き込み要求は Core ノードに転送されます。Replicant は Core ノードからのデータを完全にローカルに保持するため、読み取り操作の効率が高く、EMQX の各種データベースクエリのレイテンシ削減に寄与します。
Replicant ノードは書き込みに参加しないため、Replicant ノード数が増えても書き込みレイテンシに影響しません。これにより、数十台の Replicant ノードを持つ大規模クラスタの構築が可能です。
パフォーマンス向上のため、データ複製は独立したデータストリームに分割されます。複数の関連テーブルは同一の RLOG Shard（複製ログシャード）に割り当てられ、トランザクションは Core ノードから Replicant ノードへ順次複製されます。異なる RLOG Shard は独立しています。
Merge テーブル
Merge テーブルは Mria の特殊なテーブルで、各レコードが明確に特定の EMQX ノードに帰属します。代表例は EMQX のルーティングテーブルです。ルーティングテーブルは MQTT ブローカーにおける最重要分散データ構造であり、すべてのトピックのルーティング情報を格納します。これにより、特定トピックにパブリッシュされたメッセージをどのノードに配信すべきかを判定します。
通常のテーブルとは異なり、すべての EMQX ノード（Core でも Replicant でも）は自身のレコードを直接更新し、他ノードとの調整は行いません。その後、更新は非同期的にクラスタ全体に複製されます。つまり、各ノードは Merge テーブルに関しては Core と Replicant の両方の役割を兼ねています。
この設計の利点は以下の通りです。
- 書き込みレイテンシの低減
- Core ノードへの負荷軽減
- パーティション耐性の向上：完全にパーティション化されたネットワークでも、各ノードは少なくとも自身のルートを保持可能
ネットワークパーティションが回復した際、ノードはルーティングテーブルの内容をマージします。これが名称の由来です。
集中管理
EMQX はクラスタ内のすべてのノードを単一の管理コンソールから監視・制御できるため、集中管理が可能です。これにより、多数のデバイスやメッセージの管理が容易になります。コンソールはウェブブラウザからアクセスでき、ユーザーフレンドリーなインターフェースを提供します。任意の core タイプノードが管理用 HTTP API エンドポイントとして機能します。
オンライン設定管理機能により、クラスタ内のすべてのノードに対してノードの再起動なしに設定変更を適用可能です。これはノードの追加・削除などクラスタ設定の更新に特に有用です。
スケールの容易さ
EMQX は水平スケールが容易に行えるよう設計されています。CLI、API、またはダッシュボードからいつでもノードをクラスタに追加・削除できます。
例えば、新規ノードをクラスタに追加するには、以下のようなコマンドを実行するだけです。
emqx ctl cluster join emqx@node1.my.netここで emqx@node1.my.net はクラスタ内の既存ノードの一つです。
また、ダッシュボードからボタン操作で新規ノードの招待も可能です。
豊富な管理インターフェースにより、クラスタ管理をスクリプト化し DevOps パイプラインに組み込むことも容易です。
EMQX v5 では replica ノードはステートレス設計のため、オートスケーリンググループに配置しやすく、より良い DevOps プラクティスを実現できます。
クラスタのリバランス
新規ノードがクラスタに参加すると、初期状態は空の状態です。優れたロードバランサーがあれば、新規接続クライアントは新ノードに接続しやすくなりますが、既存クライアントは依然として旧ノードに接続し続けます。
クライアントが短期間に再接続すればクラスタはすぐにバランスしますが、再接続がない場合は長期間アンバランスな状態が続きます。
この問題に対処するため、EMQX はバージョン4.4以降で「クラスタロードリバランシング」機能を導入しました。この機能により、過負荷ノードから負荷の低いノードへセッションを移行し、クラスタの負荷を自動でリバランスできます。
「リバランス」の極端な形態は「避難（evacuation）」であり、特定ノードからすべてのセッションを移行します。ノードをクラスタから除去したい場合に有効です。
クラスタサイズ
数百万の同時接続を扱う規模では、単一マシンでの処理は不可能であり、水平スケールが必須です。
EMQX v5 の core-replica クラスタリングアーキテクチャにより、非常に大規模なクラスタを構築可能です。
ベンチマークでは、23ノードクラスタで5,000万のパブリッシャーと5,000万のワイルドカードサブスクライバーを処理しました。詳細は当社のブログ記事をご覧ください。
なぜワイルドカードかというと、ワイルドカードサブスクライブは MQTT ブローカークラスタのスケーラビリティを評価するゴールドスタンダードであり、基盤となるデータ構造とアルゴリズムに最大の負荷をかけるためです。
自動フェイルオーバー
MQTT プロトコル仕様にはセッションアフィニティの概念がありません。つまり、クライアントはクラスタ内の任意のノードに接続しても、サブスクライブしたトピックのメッセージを受信可能です。また、MQTT にはサービスディスカバリ機構もないため、クライアントはクラスタノードのアドレスを知っている必要があります。通常はクラスタ内のすべてのノードのリスト、あるいは適切なノードにルーティングできるロードバランサーを設定します。
EMQX はクラスタの前段にロードバランサーを置く設計です。ヘルスチェックエンドポイントにより、ロードバランサーはノードの健全性を検知し、クライアントを適切なノードにルーティングします。
Erlang のノード監視機構を利用し、EMQX ノードは互いの状態を監視し、不健康なノードをクラスタから自動的に除外します。
ネットワークパーティション耐性
ネットワークパーティション発生時、クラスタは複数の孤立したサブクラスタに分割され、それぞれが唯一のアクティブクラスタと誤認する「スプリットブレイン」問題が生じます。実運用クラスタはネットワークパーティションから自動復旧可能でなければなりません。
EMQX の「autoheal」機能はネットワークパーティション後のクラスタを自動的に修復します。有効化されている場合、パーティション発生後の回復時にクラスタ内ノードは以下の手順で修復処理を行います。
回復処理は Core ノードと Replicant ノードで異なります。
Core ノードの回復
- ノードはアップタイムが最長のリーダーノードにパーティション情報を報告します。
- リーダーノードはグローバルなネットスプリットビューを作成し、多数派の Core ノードの一つをコーディネーターに選出します。
- リーダーノードはコーディネーターに対し、少数派の Core ノードに再起動を指示します。
- 少数派の Core ノードは自身の Mria テーブル内容を多数派の内容で置き換えます。
Replicant ノードの回復
Replicant ノードはネットワークパーティション回復時に再起動されません。これによりクライアント接続を維持します。
代わりに以下の処理を行います。
- 少数派パーティションの Replicant は regular Mria テーブルのレプリカを再初期化します。
- ルーティングテーブルの内容をマージし、多数派と少数派のノード間でルートを再確立します。
- クライアントはグローバルセッションレジストリへの存在を再確立します。
パーティション回復時に生成されるログメッセージやアラームについては、Mria Logs and Alarms をご参照ください。
まとめ
本記事では EMQX の新しいクラスタリングアーキテクチャを紹介しました。また、スケーラビリティ、自動フェイルオーバー、ネットワークパーティション耐性など、実運用に耐える MQTT ブローカークラスタの重要な側面と、それらを実現する EMQX の仕組みについて解説しました。