Azure Event Grid MQTTでデバイス属性を知らずに属性ベース一斉配信を実現する設計パターン

大規模IoTでAzure Event Grid MQTTを使うと、「デバイスは自分の属性を知らないまま、クラウド側の属性（customerIdなど）で一斉配信したい」という要件に必ずぶつかります。本記事では、プレビュー時点の制約と回避策を整理し、現実的な設計パターンを詳しく解説します。

Azure Event Grid MQTTと属性ベース配信の前提知識

まずは、Azure Event GridのMQTTブローカー機能のキーワードをざっくり整理しておきます。

• MQTTクライアント（Client）: 実際のデバイスやアプリケーションを表す論理リソース。
• クライアント属性（Client Attributes）: customerId や type などのキー・バリュー。クライアントグループやトピックテンプレートの変数に利用される。
• クライアントグループ（Client Group）: 属性クエリ（例: attributes.customerId = 42）でクライアントを束ねる論理グループ。アクセス制御の単位。
• トピックスペース（Topic Space）: トピックテンプレートの集合。変数（${client.attributes.x} など）とMQTTワイルドカードを使って、複数トピックへのアクセス権をまとめて管理できる。
• パーミッションバインディング（Permission Binding）: 「クライアントグループが、どのトピックスペースに対して Publish / Subscribe できるか」を結びつける設定。
• ルーティング（Routing）: MQTTで受けたメッセージを Event Grid のネームスペーストピックやカスタムトピックへ転送する機能。そこからFunctionsなどで処理できる。

また、公式ドキュメントでは1つのネームスペースあたりクライアントグループは最大10個という上限が明記されています。

用語	役割	今回のシナリオでの重要ポイント
クライアント属性	クライアントを説明するキー・バリュー	customerIdなどで「どのグループに属するか」を表現したい
クライアントグループ	属性クエリで束ねるクライアント集合	属性ベースでグルーピングできるが、最大10個という上限がネック
トピックスペース	トピックテンプレートの集合	${client.attributes.customerId} などを使って属性に応じたトピック名を構成できる
パーミッションバインディング	クライアントグループとトピックスペースの紐付け	「どのトピックを購読してよいか」は制限できるが、サブスクライブフィルタ自体は自動解決されない
ルーティング	MQTTメッセージをEvent Gridトピックへ転送	クラウド側での属性フィルタリングや再配信の起点として使える

今回の要件：「デバイスは自分の属性を知らない」前提の属性ベース一斉配信

想定しているシナリオは次のようなものです。

• 多数の物理デバイスが、Azure Event GridのMQTTクライアントとして接続している。
• クラウド側のMQTTクライアントには customerId や some_attribute などの属性が付与されている。
• クラウドから「customerId = 42 のデバイスだけにコマンドを送りたい」といった属性ベースの一斉配信を行いたい。

ただし、ここで強い制約があります。

• 物理デバイスは自分の customerId を知らない・保持しない（ステートレスにしたい）。
• デバイスは「自分が属する属性に合致するトピックだけ」を受信したい。
• Event Grid側では、Topic Spaceで cloud2device/${client.attributes.customerId}/# のようなテンプレートは作れるが、サブスクライブ側のトピックフィルタには変数を使えない。
• Client Groupはネームスペースあたり10個までなので、customerId が百単位あるとスケールしない。

つまり、やりたいことは次のイメージです。

• デバイスは「汎用的な購読」を1本だけ張る（例: cloud2device/#）。
• Event GridのMQTTブローカーが、接続中クライアントの属性（customerId など）を内部で解決し、「そのクライアント向きのメッセージだけを配信」してほしい。

観点	理想	実現したい具体例
デバイスの状態	属性値を持たないステートレス	ファームウェアには customerId 等を焼き込まない
サブスクライブ	汎用トピック1本でOK	cloud2device/# に1回だけSubscribe
配信対象の決定	ブローカーが属性から自動解決	customerId = 42 のクライアントにだけ cloud2device/customer/42/… を配信
属性変更	クラウド側で書き換えれば即反映	デバイスの再設定・再接続なしにグループ変更を反映

結論：現状のEvent Grid MQTTだけではブローカー側の属性自動解決はできない

2025年時点のドキュメントおよびMicrosoft Q&Aの回答では、Event Grid MQTTブローカーに「クライアント属性を自動解決して配信対象を決める」機能は存在しないことが明言されています。

Q&Aでは、次のようなニュアンスで説明されています（要約）。

• MQTTメッセージを Event Grid のネームスペーストピックやカスタムトピックへルーティングすることはできる。
• しかし、このルーティング自体は「属性アウェア」ではなく、クライアント属性を見て配信可否を自動的に判定してくれるわけではない。
• したがって、「デバイスは汎用トピックを購読するだけで、ブローカーが属性を見て正しいメッセージだけ届ける」という挙動はサポートされていない。

また、クライアントグループとパーミッションバインディングを組み合わせれば、attributes.customerId = 42 のクライアントだけが someTopic/customerId/42 を購読できるようなアクセス制御は可能です。しかしこれは「購読できる範囲の制限」に過ぎず、「デバイスが自分の属性値を知らなくても、自動的に正しいトピックに購読が張られる」機構ではありません。

つまり、ブローカー側で魔法のように「属性ベースグループ配信＋デバイスは属性値を知らなくてよい」を完結させることは、少なくともプレビュー時点のEvent Grid MQTTではできません。

現実的な回避策・設計パターンの全体像

そこで、本記事では次のような設計パターンを紹介します。

1. Event Gridルーティング＋Azure Functionsで属性フィルタ → MQTTへ再配信
2. clientId単位の個別配信（属性→clientIdインデックス）の徹底
3. Cloud-to-Deviceトピックテンプレート活用（要件は満たしにくい）
4. Client Groups＋Permission Bindingsの活用（ただし上限に注意）
5. カスタムMQTTブローカー層を挟む（最終手段）

パターン	デバイスは属性を知らなくてよい?	スケール性	実装コスト
① ルーティング＋Functions再配信	○	◎（設計次第で大規模対応）	高（処理フロー実装が必要）
② clientId単位の個別配信	○	◎（メッセージ数は増えるが制御しやすい）	中（属性→clientIdインデックスの設計）
③ C2Dトピックテンプレート	×（デバイスが属性値を知る必要あり）	○	低〜中
④ Client Groups＋Permission Binding	△（属性は不要だが上限10件）	×（属性値が多いと厳しい）	低
⑤ カスタムMQTTブローカー	○（設計次第）	◎（自作次第）	とても高

パターン① Event Gridルーティング＋Azure Functionsで属性フィルタ → 再配信

最初のパターンは、Event Gridのルーティング機能とAzure Functionsを組み合わせて、ブローカー外で属性フィルタリング＆再ファンアウトする構成です。

構成イメージ

1. クラウドアプリは「属性ベース一斉配信リクエスト」をMQTTとして送信する。
2. Event Grid MQTTのルーティング設定で、そのメッセージをネームスペーストピックまたはカスタムトピックに転送する。
3. そのトピックに対して Event Grid サブスクリプションを張り、Azure Functions をトリガーする。
4. Function内でイベントペイロードを見て、対象の属性（customerId など）に属するclientIdの一覧をDBから取得する。
5. HTTP Publish API やサーバー側MQTTクライアントを使い、各 clientId の C2Dトピックに対して再パブリッシュする。

クラウドアプリ側の「一斉配信リクエスト」は、たとえば次のようなメッセージ構造にしておくと便利です。

{
  "type": "broadcast",
  "targetAttribute": "customerId",
  "targetValue": 42,
  "payload": {
    "command": "firmwareUpdate",
    "version": "1.2.3"
  }
}

Function はこの JSON を受け取り、targetAttribute と targetValue から「customerId = 42 のclientId一覧」を引き当てます。

HTTP Publish API を使った再パブリッシュ

サーバー側からデバイス向けの MQTT メッセージを送る場合、プレビュー機能として提供されている HTTP Publish API を使うと、HTTP POST だけで MQTT Publish を行えます。

イメージとしては以下のようなHTTPリクエストになります（実際のクエリパラメータやヘッダはドキュメントに従ってください）。

POST https://{namespace}.eventgrid.azure.net/api/mqtt/publish?api-version=2023-10-01-preview
Authorization: Bearer <token>
Content-Type: application/json

{
  "topic": "c2d/device/{clientId}/cmd",
  "qos": 1,
  "payload": {
    "command": "firmwareUpdate",
    "version": "1.2.3"
  }
}

Functions 側では、取得した clientId の配列に対してこの HTTP Publish をループしつつ、並列実行＆スロットリング（送信レート制御）を行う形になります。

メリットとデメリット

観点	メリット	デメリット
柔軟性	任意の属性条件（複数条件AND/ORなど）で配信先を決められる	ロジックがFunction側に寄るため、管理対象が増える
スケール	Functionsをスケールアウトすれば大量配信にも対応しやすい	再パブリッシュ分だけメッセージ数が増える
レイテンシ	要件次第では許容範囲内	ブローカー内完結に比べるとワンホップ分の遅延が発生
可観測性	Event Gridイベント＋Functionsログを使って配信結果を可視化しやすい	監視対象（メトリクス、ログ）が増える

「ブローカーの外に、自前のファンアウトレイヤーを作る」と割り切ることで、属性ベースの一斉配信はほぼ自由に設計可能になります。

パターン② clientId固定トピックへの個別配信（実務的ベストプラクティス）

2つ目のパターンが、今回の要件を考えたときに最も実務的で妥当な構成です。キモは「デバイスは自分の属性を知らなくても、clientId だけは知っている」ことを前提にする点です。

基本方針

• 物理デバイスは、自分の clientId に紐づく C2D トピックだけを購読する。
  • 例: c2d/<clientId>/# や device/<clientId>/cmd/#
• クラウド側では、「属性 → clientId のインデックス」をDBに保持する。
  • 例: Cosmos DB のコレクションに、{ clientId, customerId, region, ... } を保存しておく。
• 「customerId = 42 へ一斉配信したい」ときは、その属性条件でクエリし、取得した clientId のリストに対して個別にパブリッシュする。

この場合、デバイスは自分の clientId さえ知っていればよく、customerId を知らなくても問題ありません。

トピック設計例

// デバイス側が購読するトピック
c2d/{clientId}/#

// クラウド側がパブリッシュする具体例
c2d/device-0001/cmd
c2d/device-0002/cmd
...

アクセス制御の観点では、トピックスペースに次のようなテンプレートを設定しておくと、「自分のclientIdにだけアクセスできる」ように絞り込めます。

// Topic Space のトピックテンプレート例
c2d/${client.authenticationName}/#

このテンプレートを使ったトピックスペースに対して、対応するクライアントグループへSubscribe権限を付与しておけば、他人のclientIdのトピックに誤って購読することはできなくなります。

属性 → clientId インデックスの設計

Cosmos DB などを用いる場合のイメージクエリは次のようになります。

// customerId から clientId 一覧を引くクエリ例
SELECT c.clientId
FROM c
WHERE c.customerId = @customerId

属性が複数ある場合は、複合インデックスやパーティションキーを設計しておくと、大量のデバイスでも効率よく引けます。

• 単一属性配信: WHERE c.customerId = 42
• 複数属性配信: WHERE c.customerId = 42 AND c.region = "jp-east"

このインデックスは、MQTTクライアントの作成・更新イベント（MQTTClientCreatedOrUpdated など）を Event Grid 経由で受け、Function や Logic Apps で反映させる、というパターンにすると自動同期がしやすくなります。

スケール設計のポイント

clientId単位配信はメッセージ数が「一斉配信対象台数」に比例して増えるため、スケール設計が重要です。

• バッチング: clientIdを100件単位などでバッチし、Functionの1実行ごとに一定数までに抑える。
• 並列化: Functionの同時実行数を調整して、送信レートを確保しつつバックエンドのスロットルを超えないようにする。
• 再試行ポリシー: 一時的なエラー（429, 5xx）に対してはエクスポネンシャルバックオフ付きのリトライを行う。
• DLQ（デッドレターキュー）: どうしても配信できなかったメッセージはキューに避難させ、別途再処理パイプラインで対応。

このパターンで満たせる要件

• デバイスは属性値（customerId等）を一切知らなくてもよい。
• 「customerId=42のクライアントだけに配信」といった属性ベース一斉配信を実現できる。
• 権限制御を組み合わせることで、デバイス側で誤ったトピックを購読しても受信できないようにできる。

一言でいうと、「属性の話はすべてクラウド側だけで完結させ、デバイスは clientId 固有チャネルだけを見る」という割り切りがポイントです。

パターン③ Cloud-to-Deviceのトピックテンプレート活用（要件を満たしにくい）

Event Grid MQTTのトピックテンプレートでは、${client.attributes.customerId} のような変数を使って、クライアント属性に応じたトピック名を構成できます。

例えば、次のようなC2Dトピックテンプレートを定義するとします。

cloud2device/${client.attributes.customerId}/#

これにより、クラウドアプリ側は cloud2device/42/cmd にパブリッシュするだけで、「customerId=42 のクライアント」に向けたメッセージを送ることができます。

しかし問題はサブスクライブ側です。デバイスが購読フィルタとして指定できるのは、プレースホルダー解決後の具体的なトピックだけです。

• OK: cloud2device/42/#
• NG: cloud2device/${client.attributes.customerId}/#

つまり、「デバイスは自分のcustomerIdを知らなくてよい」という要件と、このパターンは相性がよくありません。デバイスが自分の「42」を知らない限り、正しいトピックを購読できないからです。

そのため、このパターンは次のようなケースに限定するとよいでしょう。

• デバイスが自分の属性値を知っていても問題ない（少数のテナント、属性がほぼ固定）
• クライアントグループとパーミッションバインディングで、トピックごとの権限をきっちり絞りたいだけのケース

パターン④ Client Groups＋Permission Bindingsの使いどころと限界

Client Groups と Permission Bindings を使うと、次のような構成が可能です。

• Client Group:
  • クエリ: attributes.customerId = 42
  • → customerId=42 のクライアントがすべてこのグループに入る。
• Topic Space:
  • トピックテンプレート: someTopic/customerId/42
• Permission Binding:
  • Client Group（customerId=42）に対して、上記Topic SpaceのSubscribe権限を付与。

これにより、someTopic/customerId/42 を購読できるのは、customerId=42 のクライアントだけ、というアクセス制御が実現できます。

しかし、ここにも大きな制約があります。

• クライアントグループはネームスペースあたり最大10個という上限がある。
• 属性値（customerIdなど）が100以上あるケースでは、「1属性値＝1クライアントグループ」の設計は成立しない。
• パーミッションバインディングは「購読できるかどうか」を制御するだけで、「どのトピックを自動的に購読するか」は決めてくれない。

観点	向いているケース	向いていないケース
テナント数	〜10テナント程度	100テナント以上
要件	「どのトピックを購読してよいか」を制限したい	「属性を知らなくても自動的に正しいトピックが購読される」
運用	権限モデルを簡潔に保ちたい	テナント追加ごとに新しいクライアントグループを増やしたくない

したがって、大量テナント・大量属性値のシナリオでは、Client Groups＋Permission Bindings は「安全側のガードレール」としてのみ使い、実際の配信ロジックは別レイヤー（パターン①や②）で設計するのが現実的です。

パターン⑤ カスタムMQTTブローカー層を挟む

どうしても「ブローカー内部で属性ベースの配信を完結させたい」という場合は、Azure Event Gridの前段に独自MQTTブローカーを置くという選択肢もあります。

• Azure上に EMQX / Mosquitto / HiveMQ などのブローカーを立てる。
• そこで属性ベースのサブスクリプションやACLを実装する（例えばデバイスIDとcustomerIdをブローカー内に持つ）。
• クラウド連携が必要なメッセージだけを Event Grid MQTT へブリッジする。

このアプローチを取ると、理論上は「好きなように属性ベース配信」を設計できますが、次のようなトレードオフがあります。

• 運用するブローカーが増える（更新・監視・スケールなど）。
• Event Grid MQTT固有の利点（他Azureサービスとの統合、Event Gridルーティングなど）がやや薄まる。
• 障害ポイントが増える。

そのため、本当にEvent Grid MQTTだけでは要件を満たせない場合の「最後の手段」として検討するのがよいでしょう。

設計・運用上の実務ノウハウ

属性変更とインデックス更新

属性 → clientId インデックスを持つ構成では、「属性が変わった瞬間にどのようにインデックスを更新するか」が重要です。

• Event Grid Namespace から出る MQTT クライアント関連のイベント（作成・更新・削除・セッション接続/切断）をサブスクライブする。
• これをトリガーにして Cosmos DB やキャッシュ（Redisなど）のインデックスを更新する。
• 更新頻度が高い場合は、「直近だけキャッシュ、フルはCosmos DBクエリ」の二段構成も検討する。

スロットリングとバックプレッシャー

大量のデバイスに一斉配信する場合、配信レートを制御しないと、Event Gridや下流サービスのスロットル（429）を踏みやすくなります。

• Azure Functions の同時実行数・バッチサイズを制御する。
• HTTP Publish API を使う場合は、1秒あたりのリクエスト数をメトリクス化し、上限を決める。
• Rate Limit を超えたら、キューに戻して後続処理に回す（バックプレッシャー）。

セキュリティと誤配信防止

属性ベース配信でもっとも怖いのは「誤ったグループにコマンドを送ってしまう」パターンです。このリスクを小さくするには、二重三重の防御を入れておくと安心です。

• Permission Bindings で、デバイスが購読できるトピックを厳格に制限する。
• トピック命名規則を明確にして、「customerId を含まないC2Dトピック」「clientId専用C2Dトピック」など役割を分ける。
• クラウド側の送信コードでは、「本当にこの属性値で合っているか？」をチェックするバリデーションレイヤーを用意する。

可観測性：配信数・失敗数・遅延のモニタリング

Event Grid Namespace は、MQTTクライアントの接続・切断イベントや、イベント配信のメトリクスを提供しています。 これらと Azure Monitor / Application Insights を組み合わせることで、次のような指標を可視化できます。

• 一斉配信リクエスト数
• 実際に送信されたMQTTメッセージ数
• 成功・失敗・リトライ回数
• 完了までのレイテンシ（配信リクエストから最後のメッセージ送信完了までの時間）

これらをダッシュボード化しておくと、「どの属性グループへの配信が重いのか」「どのタイミングでスロットルに引っかかっているのか」といったボトルネックの特定がしやすくなります。

ステップバイステップ例：clientId固定トピック＋属性→clientId解決で一斉配信

最後に、実際に構成を作るときのステップをまとめます。ここではパターン②（clientId固定トピック＋属性→clientId解決）をベースにします。

ステップ1: トピック設計

• デバイス→クラウド（テレメトリ）:
  • telemetry/{clientId}/...
• クラウド→デバイス（コマンド）:
  • c2d/{clientId}/cmd
  • c2d/{clientId}/config

デバイスは起動時に c2d/{clientId}/# をサブスクライブしておくだけでOKです。

ステップ2: MQTTリソース（クライアント・トピックスペース・権限）の定義

• クライアントの authenticationName に clientId をそのまま使う。
• Topic Space に c2d/${client.authenticationName}/# のテンプレートを定義する。
• そのTopic Spaceに、すべてのデバイスを含むClient Group（あるいは$all）を、Subscribe権限でバインドする。

ステップ3: 属性ストアとインデックスの構築

• Cosmos DB に次のようなドキュメントを保存する:
  • { clientId, customerId, region, model, ... }
• パーティションキーやインデックスを customerId / region などのクエリしやすいキーに合わせて設計する。
• Event Grid の MQTT クライアントイベントをトリガーに、属性ストアを自動同期する。

ステップ4: 一斉配信フローの実装

1. クラウドアプリから「属性ベース一斉配信リクエスト」が飛んでくる（HTTP APIやキューなど）。
2. Azure Functions（またはLogic Apps）がこのリクエストを受ける。
3. 属性条件に基づいて Cosmos DB から clientId 一覧を取得する。
4. 各 clientId に対して、HTTP Publish API で c2d/{clientId}/cmd にメッセージをPublishする。
5. 送信結果をログ＆メトリクスに記録し、失敗したものはリトライまたはDLQへ送る。

ステップ5: 監視と運用

• 配信リクエスト数・実際のメッセージ数・失敗数・レイテンシをダッシュボード化する。
• 429や5xxが増えた場合にアラートを上げる。
• クライアントの接続/切断イベントの異常（特定グループだけ切断が多い等）を検知する。

まとめ：ブローカーの外で「属性→対象デバイス解決」をやるのが現実解

本記事で見てきた通り、2025年時点のAzure Event Grid MQTTには、

• クライアント属性をブローカー内部で自動解決し、
• デバイスが属性値を知らなくても正しいトピックだけを受信する、

といったフルマネージドな属性ベース一斉配信機能は存在しません

そのため、

• ブローカーは「セキュアなMQTTハブ」として割り切り、
• 属性ベースの対象解決はクラウドアプリ（Functionsなど）で行い、
• clientId 固定C2Dトピックへの個別ファンアウトで要件を満たす、

という構成が、現状もっとも現実的でスケールしやすい設計と言えます。

将来、Event Grid MQTTに属性ベースグループ配信のネイティブ機能が追加される可能性はありますが、少なくとも現行プレビュー仕様では、「属性→対象デバイス解決はアプリケーション側の責務」として設計しておくのが安全です。