TLDR: 理にかなっているように見えますが、CreateConsumerGroupIfNotExists で異なる名前を使用して 2 つのコンシューマー グループを作成するだけです。
コンシューマー グループは主に概念であるため、実際にどのように機能するかは、サブスクライバーがどのように実装されているかによって異なります。ご存知のように、概念的には、各グループがすべてのメッセージを受信し、理想的な (起こり得ない) 状況では、各メッセージを 1 回消費するように、それらは一緒に作業するサブスクライバーのグループです。これは、各コンシューマー グループが「同じワーカー ロールの複数のインスタンスによってすべてのパーティションが処理される」ことを意味します。あなたはこれを求めている。
これはさまざまな方法で実装できます。Microsoft は、Event Hubs からのメッセージを直接消費する 2 つの方法と、おそらく 2 つの直接的な方法の上に構築された Streaming Analytics などを使用するオプションを提供しています。最初の方法はEvent Hub Receiverで、2 番目のより高いレベルはEvent Processor Hostです。
私はイベント ハブ レシーバーを直接使用したことがないため、この特定のコメントは、これらの種類のシステムがどのように機能するかについての理論とドキュメントからの推測に基づいています。これらを使用する場合は、オフセットのすべての調整とコミットを自分で行う必要があります (できる!)。これは、計算された集計と同じトランザクションでオフセットをトランザクション DB に書き込むなどのいくつかのシナリオで利点があります。これらの低レベル受信機の使用、同じ Azure コンシューマー グループを使用する異なる論理コンシューマー グループを持つことは、おそらく特に問題になるべきではありません (実用的なアドバイスではありません)。
より有用な情報に移りましょう。 EventProcessorHostsはおそらく EventHubReceivers の上に構築されています。それらはより高いレベルのものであり、複数のマシンが論理的なコンシューマー グループとして連携できるようにするためのサポートがあります。以下に、 EventProcessorHostを作成するコードのスニペットを軽く編集して、いくつかの選択肢を説明するコメントを残しました。
//We need an identifier for the lease. It must be unique across concurrently
//running instances of the program. There are three main options for this. The
//first is a static value from a config file. The second is the machine's NETBIOS
//name ie System.Environment.MachineName. The third is a random value unique per run which
//we have chosen here, if our VMs have very weak randomness bad things may happen.
string hostName = Guid.NewGuid().ToString();
//It's not clear if we want this here long term or if we prefer that the Consumer
//Groups be created out of band. Nor are there necessarily good tools to discover
//existing consumer groups.
NamespaceManager namespaceManager =
NamespaceManager.CreateFromConnectionString(eventHubConnectionString);
EventHubDescription ehd = namespaceManager.GetEventHub(eventHubPath);
namespaceManager.CreateConsumerGroupIfNotExists(ehd.Path, consumerGroupName);
host = new EventProcessorHost(hostName, eventHubPath, consumerGroupName,
eventHubConnectionString, storageConnectionString, leaseContainerName);
//Call something like this when you want it to start
host.RegisterEventProcessorFactoryAsync(factory)
存在しない場合は新しいコンシューマー グループを作成するように Azure に指示したことに気付くでしょう。存在しない場合は、素敵なエラー メッセージが表示されます。EventProcessorHost の調整 (およびおそらくコミット) が適切に機能するために、インスタンス間で同じにする必要があるストレージ接続文字列が含まれていないため、これの目的が正直にわかりません。
ここでは、11 月に実験していたコンシューマー グループからのリースとおそらくオフセットのAzure Storage Explorerからの画像を提供しました。私は testhub と testhub-testcg コンテナーを持っていますが、これは手動で名前を付けているためです。それらが同じコンテナにある場合、「$Default/0」と「testcg/0」のようなものになります。
ご覧のとおり、パーティションごとに 1 つの BLOB があります。私の推測では、これらのブロブは 2 つの目的で使用されます。これらの 1 つ目は、インスタンス間でパーティションを分散するための BLOB リースです。ここを参照してください。2 つ目は、コミットされたパーティション内のオフセットを格納することです。
コンシューマー グループにデータがプッシュされるのではなく、コンシューマー インスタンスがストレージ システムに対して 1 つのパーティション内のオフセットでデータを要求しています。EventProcessorHosts は、各パーティションが一度に 1 つのコンシューマーによってのみ読み取られ、論理コンシューマー グループが各パーティションで行った進行状況が忘れられない論理コンシューマー グループを持つ優れた高レベルの方法です。
パーティションごとのスループットが測定されるため、イングレスを最大限に活用している場合は、すべてが高速な 2 つの論理コンシューマーしか持てないことに注意してください。そのため、次のことができる十分なパーティションとスループット ユニットがあることを確認する必要があります。
- 送信したすべてのデータを読み取ります。
- 問題が原因で数時間遅れた場合は、24 時間の保持期間内に追いつくことができます。
結論として、消費者グループは必要なものです。特定のコンシューマー グループを使用する例は適切です。各論理コンシューマー グループ内では、Azure コンシューマー グループに同じ名前を使用し、異なる論理コンシューマー グループに異なる名前を使用させます。
私はまだ Azure Stream Analytics を使用していませんが、少なくともプレビュー リリース中は、既定のコンシューマー グループに限定されます。そのため、既定のコンシューマー グループを他の目的で使用しないでください。Azure Stream Analytics の 2 つの別々のロットが必要な場合は、厄介なことをしなければならない場合があります。でも設定は簡単！
