MQTT 共有サブスクライブ
EMQX は MQTT の共有サブスクライブ機能を実装しています。共有サブスクライブとは、複数のサブスクライバー間で負荷分散を実現するためのサブスクライブモードです。クライアントは複数のサブスクライブグループに分けられ、メッセージはすべてのサブスクライブグループに転送されますが、各グループ内のクライアントのうち一つだけがメッセージを受信します。複数のサブスクライバーに対して共有サブスクライブを有効にするには、元のトピックにプレフィックス $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のうちの一つだけがmsg1を受信します。s4とs5のうちの一つだけが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 | グループ内のサブスクライバーをランダムに選択してメッセージを配信します。全体的に均等な分配を提供しますが、決定論的ではありません。 |
round_robin (デフォルト) | 各共有サブスクライブグループ内で順番にメッセージを配信します。EMQX は各パブリッシャークライアントごとに別々のラウンドロビン位置を保持するため、異なるパブリッシャーからのメッセージが同じサブスクライバーに連続して配信されることがあります。 |
round_robin_per_group | round_robin と似ていますが、ラウンドロビンの進行は各ノードで独立して追跡されます。これにより、クラスター内の異なるノードからパブリッシュされたメッセージが同じサブスクライバーに配信されることがあります。各ノードが別々のパブリッシャーロードを処理するクラスター展開に有用です。 |
sticky | サブスクライバーが切断またはセッション終了するまで、同じサブスクライバーに継続的にメッセージを配信します。この戦略は、可能な限りメッセージフローを単一サブスクライバーに保つことで、不要な再処理やクライアント間の状態共有を回避します。 初期のサブスクライバーは mqtt.shared_subscription_initial_sticky_pick 設定で決定されます。 |
local | メッセージが処理されるノードに接続されているサブスクライバーへの配信を優先します。ローカルサブスクライバーが存在しない場合、EMQX はクラスター内のサブスクライバーをランダムに選択します。これによりノード間のトラフィックとレイテンシを削減します。 |
hash_clientid | パブリッシャーの Client 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接続を一つずつ選択し、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 は異なるグループに同時にメッセージを転送しますが、同じグループ内のサブスクライバーのうち一つだけがメッセージを受信します。
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新しいクライアントを使って、元のトピック
t/1にペイロード1、2、3、4の4つのメッセージをパブリッシュします：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）は同時にメッセージを受信します。
- 同じグループ内のサブスクライバーのうち一つだけがメッセージを受信します。