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つのDB管理システムは性質が大きく異なり、ネットワークパーティション時の保証も異なります。本ドキュメントでは主に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ノード数に抑え、残りはReplicant役割を割り当てることが推奨されます。一般的にクラスタ内のCoreノード数は3台です。
Replicantノードはトランザクション処理に直接関与せず、Coreノードに接続してデータ更新を受動的に複製します。Replicantは書き込み操作を行わず、書き込み要求はCoreノードに転送されます。ReplicantはCoreノードからのデータを完全にローカルに保持するため、読み取り操作の効率が高く、EMQXの各種DBクエリのレイテンシ低減に寄与します。
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以降）は「クラスタロードリバランシング」機能を導入しました。この機能により、過負荷ノードから低負荷ノードへセッションを自動的に移行し、クラスタの負荷をリバランスします。
「リバランス」の極端な例が「エバキュエーション（避難）」で、指定ノードからすべてのセッションを移行します。ノードをクラスタから除外する際に有用です。
クラスタサイズ
数百万の同時接続規模では、単一マシンでの処理は不可能であり、水平スケールが必須です。
EMQX v5のcore-replicaクラスタリングアーキテクチャにより、はるかに大規模なクラスタの構築が可能です。
ベンチマークでは、23ノードクラスタで5,000万のパブリッシャーと5,000万のワイルドカードサブスクライバーを同時に処理しました。詳細はブログ記事をご覧ください。
なぜワイルドカードかというと、ワイルドカードサブスクライブはMQTTブローカークラスタのスケーラビリティを評価するゴールドスタンダードであり、基盤となるデータ構造とアルゴリズムに最大の負荷をかけるためです。
自動フェイルオーバー
MQTTプロトコル仕様にはセッションアフィニティの概念がありません。つまり、クライアントはクラスタ内の任意のノードに接続しても、サブスクライブしたトピックのメッセージを受信可能です。またMQTTにはサービスディスカバリ機構もないため、クライアントはクラスタノードのアドレスを知っている必要があります。通常、クライアントはクラスタ内のすべてのノード一覧、または適切なノードにルーティング可能なロードバランサーを設定されます。
EMQXはクラスタの前段にロードバランサーを置く設計です。ヘルスチェックエンドポイントにより、ロードバランサーはクラスタノードの健全性を検知し、クライアントを適切なノードにルーティングします。
Erlangのノード監視機構を用いて、EMQXノードは互いの健全性を監視し、不健康なノードを自動的にクラスタから除外します。
ネットワークパーティション耐性
ネットワークパーティションが発生すると、クラスタは複数の孤立したサブクラスタに分割され、それぞれが唯一のアクティブクラスタと誤認する「スプリットブレイン」問題が生じます。運用中のクラスタはネットワークパーティションから自動的に復旧可能でなければなりません。
EMQXの「autoheal」機能はネットワークパーティション後のクラスタを自動的に修復します。有効化すると、パーティション発生後の回復時にクラスタ内ノードは以下の手順で修復を行います。
復旧処理はCoreノードとReplicantノードで異なります。
Coreノードの復旧
- ノードは最も長いアップタイムを持つリーダーノードにパーティション情報を報告します。
- リーダーノードはグローバルなネットスプリットビューを作成し、多数派のCoreノードの1つをコーディネーターに選出します。
- リーダーノードはコーディネーターに少数派のCoreノードを再起動させるよう指示します。
- 少数派Coreノードは自身のMriaテーブル内容を多数派の内容で置き換えます。
Replicantノードの復旧
Replicantノードはネットワークパーティション修復時に再起動されません。これによりクライアント接続を維持可能です。
代わりに以下の処理を行います。
- 少数派パーティションのReplicantはregular Mriaテーブルのレプリカを再初期化します。
- ルーティングテーブルの内容をマージし、多数派と少数派のノード間でルートを再確立します。
- クライアントはグローバルセッションレジストリへの存在を再確立します。
パーティション復旧時に生成されるログメッセージやアラームについてはMrIA Logs and Alarmsをご参照ください。
まとめ
本記事ではEMQXの新しいクラスタリングアーキテクチャを紹介しました。また、スケーラビリティ、自動フェイルオーバー、ネットワークパーティション耐性など、実運用に耐えるMQTTブローカークラスタの重要な側面と、それらを実現するEMQXの仕組みについて解説しました。