MQTT 共有サブスクライブ
EMQX は MQTT の共有サブスクライブ機能を実装しています。共有サブスクライブとは、複数のサブスクライバー間で負荷分散を実現するためのサブスクライブモードです。クライアントは複数のサブスクライブグループに分けられ、メッセージはすべてのサブスクライブグループに転送されますが、各グループ内のクライアントのうち一度に受信するのは1つだけです。複数のサブスクライバーで共有サブスクライブを有効にするには、元のトピックの前にプレフィックス $share/<group-name>/ を追加します。
共有サブスクライブのプレフィックス形式の例は以下の通りです。
| プレフィックス形式 | 例 | プレフィックス | 実際のトピック名 |
|---|---|---|---|
| グループの共有サブスクライブ | $share/abc/t/1 | $share/abc/ | t/1 |
クライアントツールを使って EMQX に接続し、このメッセージングサービスを試すことができます。本節では共有サブスクライブの動作を紹介し、MQTTX Desktop と MQTTX CLI を使ってクライアントをシミュレートし、共有サブスクライブ機能を試す方法を説明します。
グループの共有サブスクライブ
元のトピックにプレフィックス $share/<group-name> を付けることで、サブスクライバーのグループに対して共有サブスクライブを有効にできます。グループ名は任意の文字列を指定可能です。EMQX は異なるグループに同時にメッセージを転送し、同じグループに属するサブスクライバーは負荷分散された形でメッセージを受信します。
例えば、サブスクライバー s1、s2、s3 がグループ g1 のメンバーで、サブスクライバー s4 と s5 がグループ g2 のメンバーであり、すべてのサブスクライバーが元のトピック t1 をサブスクライブしている場合、共有サブスクライブのトピックは $share/g1/t1 と $share/g2/t1 となります。EMQX が元のトピック t1 にメッセージ msg1 をパブリッシュすると：
- EMQX は
msg1をグループg1とg2の両方に送信します。 s1、s2、s3のうちのいずれか1つだけがmsg1を受信します。s4とs5のうちのいずれか1つだけがmsg1を受信します。
廃止予定の $queue/ プレフィックス
EMQX はグループ名なしの共有サブスクライブとして $queue/ プレフィックスも受け付けていますが、この使用法は廃止予定です。メッセージキュー が有効な場合、EMQX は $queue/ プレフィックスをキューサブスクライブ用に予約するため、同じトピックフィルターは共有サブスクライブを作成しなくなります。代わりに $share/<group-name>/ を使用してください。例えば $queue/t/1 のサブスクライブは $share/$queue/t/1 と同等です。
共有サブスクライブとセッション
クライアントがパーシステントセッションを持ち、共有サブスクライブにサブスクライブしている場合、クライアントが切断している間もセッションは共有サブスクライブトピックにパブリッシュされたメッセージを受信し続けます。クライアントが長時間切断状態にあり、メッセージのパブリッシュ頻度が高い場合、セッション状態内の内部メッセージキューがオーバーフローする可能性があります。この問題を避けるため、共有サブスクライブではクリーンセッション（clean_session=true）を使用することを推奨します。クリーンセッションはクライアント切断後すぐに期限切れになります。
MQTT v5 を使用するクライアントでは、短いセッション有効期間（0でない場合）を設定することが望ましいです。これによりクライアントは一時的に切断して再接続し、切断期間中にパブリッシュされたメッセージを受信できます。セッションが期限切れになると、送信キュー内の QoS1 および QoS2 メッセージ、またはインフライトキュー内の QoS1 メッセージは同じグループ内の他のセッションに再配信されます。最後のセッションが期限切れになると、すべての保留中メッセージは破棄されます。
パーシステントセッションの詳細は MQTT Persistent Session and Clean Session Explained をご参照ください。
共有サブスクライブ戦略の設定
EMQX は共有サブスクライブグループ内のサブスクライバーへのメッセージ配信方法を細かく制御できます。この動作は共有サブスクライブ戦略によって定義され、各メッセージを受信するサブスクライバーを選択するアルゴリズムを決定します。この戦略を調整することで、異なるワークロード、クライアントの配置パターン、クラスターのトポロジーに合わせてメッセージフローを最適化できます。
共有サブスクライブ配信戦略
共有サブスクライブ配信戦略は、EMQX が共有サブスクライブグループ内のサブスクライバー間でメッセージをどのように分配するかを決定します。
mqtt.shared_subscription_strategy オプションで設定可能です。
| 戦略 | 説明 |
|---|---|
random | グループ内のサブスクライバーをランダムに1つ選択してメッセージを配信します。全体として均等に分配されますが、決定論的ではありません。 |
round_robin（デフォルト） | メッセージを各共有サブスクライブグループ内のサブスクライバーに順番に配信します。EMQX は各パブリッシャークライアントごとに独立したラウンドロビンの位置を管理するため、異なるパブリッシャーからのメッセージが同じサブスクライバーに連続して配信されることがあります。 |
round_robin_per_group | round_robin に似ていますが、ラウンドロビンの進行状況は各ノードで独立して追跡されます。つまり、クラスター内の異なるノードからパブリッシュされたメッセージが同じサブスクライバーに配信される可能性があります。各ノードが別々のパブリッシャーロードを処理するクラスター環境に適しています。 |
sticky | サブスクライバーが切断またはセッション終了するまで、同じサブスクライバーに継続的にメッセージを配信します。この戦略は、可能な限りメッセージフローを単一のサブスクライバーに保つことで、不要な再処理やクライアント間の状態共有を回避します。 初期サブスクライバーは mqtt.shared_subscription_initial_sticky_pick 設定で決定されます。 |
local | メッセージが処理されるノードに接続されているサブスクライバーへの配信を優先します。ローカルサブスクライバーが存在しない場合は、クラスター内のサブスクライバーをランダムに選択します。これによりノード間のトラフィックとレイテンシが低減されます。 |
hash_clientid | パブリッシャーのクライアントIDのハッシュを使用して、そのパブリッシャーからのすべてのメッセージを同じサブスクライバーに一貫してルーティングします。クライアント単位のデータストリームに対して決定論的なルーティングを提供します。 |
hash_topic | パブリッシュされたトピックのハッシュを使用して、同じトピック名のすべてのメッセージを同じサブスクライバーにルーティングします。トピックベースのメッセージシャーディングや状態を持つ処理に有用です。 |
EMQX はラウンドロビンの位置やスティッキーサブスクライバーの割り当てなどの配信状態をパブリッシャー接続ごとに追跡します。パブリッシャークライアントが切断して再接続すると、この状態はリセットされ再初期化されます。
初期スティッキーピック
共有サブスクライブ戦略が sticky に設定されている場合、EMQX は mqtt.shared_subscription_initial_sticky_pick に基づいてメッセージを受信する初期サブスクライバーを決定します。
この設定は、EMQX がローカルサブスクライバーを優先するか、決定論的ルーティングのためにハッシュを使うか、ランダムに選択するかを制御します。
ダッシュボードでの配信戦略設定
EMQX ダッシュボードで以下の手順で設定できます。
- 管理 -> MQTT 設定 -> 一般 に移動します。
- 許可された共有サブスクライブ が有効になっていることを確認します。
- ドロップダウンリストから希望の 共有サブスクライブ戦略 を選択します。デフォルトは
round_robinです。 - 選択した戦略が
stickyの場合、適切な 共有サブスクライブ初期スティッキーピック 方法を選択します。デフォルトはrandomです。 - 変更を保存 をクリックします。
これらの設定は即時に反映され、クライアント側の設定変更なしに EMQX のメッセージ配信ロジックを調整できます。
MQTTX Desktop で共有サブスクライブを試す
以下の手順は、元のトピックに $share プレフィックスを付けて、異なるグループのサブスクライバーが同じトピックの共有サブスクライブを行い、共有サブスクライブからメッセージを受信する方法を示します。
このデモでは、1つのクライアント接続 demo をパブリッシャーとして作成し、トピック t/1 にメッセージをパブリッシュします。次に、4つのクライアント接続をサブスクライバーとして作成し、それぞれ Subscriber1、Subscriber2、Subscriber3、Subscriber4 とします。サブスクライバーはグループ a と b に分けられ、両グループがトピック t/1 をサブスクライブします。
EMQX と MQTTX Desktop を起動します。New Connection をクリックしてパブリッシャー用のクライアント接続を作成します。
- Name フィールドに
Demoと入力します。 - Host にローカルホスト
127.0.0.1を入力します（このデモの例として）。 - その他の設定はデフォルトのままにして Connect をクリックします。
TIP
MQTT 接続の詳細な作成方法は MQTTX Desktop をご参照ください。
- Name フィールドに
New Connection をクリックして、4つの新しい接続をサブスクライバーとして作成します。Name をそれぞれ
Subscriber1、Subscriber2、Subscriber3、Subscriber4に設定します。Connections ペインで
Subscriber接続を1つずつ選択し、New Subscription をクリックして各サブスクライバーに共有サブスクライブを作成します。以下のルールに従って Topic テキストボックスに正しいトピックを入力します。複数のサブスクライバーでグループを形成するには、サブスクライブするトピック
t/1の前にグループ名{group}を追加します。すべてが同じトピックをサブスクライブするようにするには、グループ名の前にプレフィックス$shareを追加します。New Subscription ウィンドウで：
Subscriber1とSubscriber2の Topic を$share/a/t/1に設定します。Subscriber3とSubscriber4の Topic を$share/b/t/1に設定します。
これらの例のトピックでは：
- プレフィックス
$shareは共有サブスクライブであることを示します。 {group}はaとbですが、任意の名前に変更可能です。t/1は元のトピック名です。
その他の設定はデフォルトのままにして Confirm をクリックします。
先ほど作成した
Demo接続をクリックします。トピック
t/1でメッセージを送信します。グループaのクライアントSubscriber1とグループbのクライアントSubscriber4がメッセージを受信するはずです。
同じメッセージを再度送信します。グループ
aのクライアントSubscriber2とグループbのクライアントSubscriber3がメッセージを受信するはずです。
TIP
共有サブスクライブのメッセージがパブリッシュされると、EMQX は異なるグループに同時にメッセージを転送しますが、同じグループ内のサブスクライバーのうち1つだけがメッセージを受信します。
MQTTX CLI で共有サブスクライブを試す
4つのサブスクライバーを2つのグループに分けてトピック
t/1をサブスクライブします：bash# クライアント A と B はトピック `$share/my_group1/t/1` をサブスクライブ mqttx sub -t '$share/my_group1/t/1' -h 'localhost' -p 1883 ## クライアント C と D はトピック `$share/my_group2/t/1` をサブスクライブ mqttx sub -t '$share/my_group2/t/1' -h 'localhost' -p 1883新しいクライアントでペイロード
1、2、3、4の4つのメッセージを元のトピックt/1にパブリッシュします：bashmqttx pub -t 't/1' -m '1' -h 'localhost' -p 1883 mqttx pub -t 't/1' -m '2' -h 'localhost' -p 1883 mqttx pub -t 't/1' -m '3' -h 'localhost' -p 1883 mqttx pub -t 't/1' -m '4' -h 'localhost' -p 1883各サブスクライブグループ内のクライアントが受信するメッセージを確認します：
- サブスクライブグループ1（A と B）およびサブスクライブグループ2（C と D）は同時にメッセージを受信します。
- 同じグループ内のサブスクライバーのうち、1つだけがメッセージを受信します。