Windows Azureキューのスケーリング

この記事では、スケーラブルで高性能で費用対効果の高いWindows Azureキューベースのメッセージングソリューションを構築するためのガイドラインを提供します。 このドキュメントは、Windows Azureキューを使用するクラウドソリューションの設計者および開発者を対象としています。

はじめに

従来のキューベースのメッセージングソリューションは、 メッセージキューと呼ばれるリポジトリを使用します。 これは、非同期交換メカニズムを使用して参加者が受信または送信したデータのリポジトリです。

キューベースのデータ交換は、分散コンピューティング環境で幅広いユニバーサルシナリオをサポートする、フォールトトレラントでスケーラブルなメッセージングアーキテクチャを作成するための強固な基盤です。 目の前のタスク（大量の仕事のスケジューリングや信頼性の高いメッセージング）に関係なく、メッセージキューテクノロジは、非同期データ転送の要件に応じた最高の情報交換機能を提供します。

このドキュメントでは、Windows Azureプラットフォームの機能と、設計パターンを使用して、合理化された低コストのキューベースのメッセージングシステムを作成する方法について説明します。 このドキュメントには、Windows Azureプラットフォームの最新のソリューションでキューベースの対話ソフトウェアを実装するための主な方法の詳細な概要と、生産性の向上、スケーラビリティの向上、運用コストの削減に関する推奨事項が含まれています。

スクリプト

明確にするために、実際のシナリオを次のように一般化します。 SaaSソリューションプロバイダーは、Windows Azureアプリケーションとして実装された新しい課金システムをコミッションし、多数の顧客トランザクションを処理する企業のニーズに応えています。 このソリューションは、ワークロードをクラウドに転送し、Windows Azureインフラストラクチャの柔軟性を使用して複雑な計算を実行することに基づいています。

ローカル統合インフラストラクチャ要素は、Windows Azureクラウド環境でホストされるサービスによる後続の処理のために、定期的に毎日大量のトランザクションを統合およびディスパッチします。 転送されるトランザクションの量は、1つのパッケージ内で数千から数十万まで変化し、1日の合計量は数百万トランザクションに達する可能性があります。 このソリューションは、データ処理の最大遅延を保証するという点で、サービスレベル契約（SLA）の要件を満たす必要があります。

ソリューションアーキテクチャは、map-reduceと呼ばれる分散システムテンプレートに基づいており、Windows Azureキューストレージを使用して作業をディスパッチする多くのクラウドレベルのワーカーロールで構成されています。 トランザクションパッケージは、 Process Initiator Workerロールによって受け入れられます。 次に、ワークロードを分散するためにWindows Azureキューに渡される小さな作業タスクに分割されます。

ワークロードは、キューから作業操作を抽出し、計算手順を実行するProcessing Workerロールの多数のインスタンスによって処理されます。 これらのハンドラーインスタンスは、マルチスレッドキューリスナーを使用して、パフォーマンスを最大化するためにデータを並列処理します。

処理された作業項目は専用のキューにリダイレクトされ、そこからデータマイニングとレポートのためにウェアハウスでの集約と長期保管のためにプロセスコントローラーワーカーロールのインスタンスが取得されます。

ソリューションのアーキテクチャは次のとおりです。


上の図は、大規模または複雑なコンピューティング負荷をスケーリングするために使用される一般的なアーキテクチャの例を示しています。 このアーキテクチャで実装されたキューベースのメッセージングテンプレートは、キューを使用して相互に通信する必要がある多くのWindows Azureアプリケーションおよびサービスの典型です。 これにより、キューベースのメッセージングに使用される主要なコンポーネントを研究するための標準的なアプローチが可能になります。

キューメッセージングの基本

分散コンポーネント間のデータ交換をサポートする標準メッセージングソリューションには、メッセージをキューに入れるパブリッシャーと、これらのメッセージを受信する1人以上のサブスクライバーが含まれます。 ほとんどの場合、 キューリスナーと呼ばれることもあるサブスクライバーは、ユーザーのスケジュールに従って連続的に実行されるか、オンデマンドで実行されるシングルスレッドまたはマルチスレッドアプリケーションとして実装されます。

より高いレベルでは、キューのリスナーがキューに格納されたメッセージを受信できるようにする2つの主要なディスパッチメカニズムが使用されます。

• ポーリング（クエリベースモデル-ポーリング） ：リスナーはキューを監視し、定期的に新しいメッセージをチェックします。 キューが空の場合、リスナーはキューのクエリを続行し、定期的にスリープ状態に変わります。
• 切り替え（強制データ送信に基づくプッシュモデル） ：リスナーは、トリガーが関連付けられているイベント（パブリッシャーまたはキューマネージャーによる）にサブスクライブします。これは、メッセージがキューに送信されたときにトリガーされます。 リスナーはメッセージの処理を開始できるため、新しいデータの可用性を判断するためにキューを再クエリする必要はありません。

各メカニズムの実際の実装には独自の特性があることを強調する必要があります。 たとえば、ポーリングはブロッキングまたはノンブロッキングのいずれかです。 ロックは、新しいメッセージがキューに表示されるまで（またはタイムアウトが期限切れになるまで）要求を保留にしますが、キューが空の場合は非ブロック要求がすぐに実行されます。 スイッチングモデルを使用すると、最初のメッセージが空のキューに到着するたびに、またはキューの深さが特定の値に達するたびに、通知をキューリスナーに送信できます。

ご注意 Windows Azure Queue Service APIでサポートされているデキュー操作は非ブロッキングです。 つまり、メッセージキューが空の場合、 GetMessageまたはGetMessages APIメソッドはすぐに終了します。 対照的に、Windows Azure統合バスによって提供されるDurable Message Buffers （DMB）は、メッセージがDMBキューに到着するか、指定されたタイムアウトが期限切れになるまで呼び出しスレッドがブロックされるブロックメッセージ取得操作を使用します。

Windows Azureキューリスナーのソフトウェア実装に対する次の最も一般的なアプローチは区別できます。

1. リスナーはアプリケーションコンポーネントとして実装され、そのインスタンスは作業ロールインスタンスの一部として作成および実行されます。
2. ロールリスナコンポーネントのライフサイクルは、多くの場合、ホストされたロールのインスタンスの実行時間に関連付けられています。
3. 主な処理ロジックは、メッセージがキューから削除されて処理のために送信されるループです。
4. 受信したメッセージのキューが空の場合、リスナーはスリープモードに入ります。スリープモードの持続時間は、システムを停止するアルゴリズムによって決定され、アプリケーションによって異なります。
5. メッセージを受信するためのプログラムサイクルが実行されます。 リスナーがループを終了して作業を完了するための通知を受け取るまで、キューに問い合わせが行われます。

次のフローチャートは、Windows Azure環境で組み込みのポーリングメカニズムを使用してキューリスナーを実装するための標準ロジックを示しています。

ご注意 たとえば、中央のキューマネージャー（ブローカー）を必要とする、より複雑な決定パターンの説明は、このドキュメントの範囲外です。

ポーリングメカニズムと組み合わせてクラシックキューリスナーを使用することは、最良の選択ではありません。 Windows Azureの価格モデルは、キューがいっぱいであるかどうかに関係なく、キュー内のアプリケーションリクエストの数を考慮して、リポジトリ内のトランザクション数をカウントすることに基づいています。 次のセクションでは、Windows Azureキューイングメッセージングシステムの実装の生産性を最大化し、コストを最小化する方法について説明します。

ソリューションのコストを削減し、パフォーマンスとスケーラビリティを確保するための推奨事項

このセクションでは、ターンキーソリューションのコストを削減するだけでなく、生産性とスケーラビリティを向上させる設計方法について説明します。

システム実装テンプレートは、次の目標を確実に達成できる場合にのみ、より効果的なソリューションと呼ぶことができます。

• 有用な作業を実行しないストレージとのデータ交換のためのトランザクションの拒否による運用コストの削減 。
• キュー内の新しいメッセージをチェックする際のポーリング間隔の使用に関連する不要な遅延の除去 。
• 変化するワークロードにコンピューティングパワーを適合させることによる動的なスケーリング （拡張および縮小）。

実装テンプレートは、システムを複雑にすることなくこれらのタスクを実行する必要があります。そうしないと、実装の利点が無効になります。

データをストレージと交換する際のトランザクションコストを最適化するための推奨事項

Windows Azureプラットフォームに展開されたソリューションの総所有コスト（TCO）と投資収益率（ROI）の指標を評価する場合、TCO式の最も重要な変数の1つは、ストレージとのデータ交換中のトランザクションの量です。 Windows Azureキューとデータを交換するためのトランザクションの数を減らすと、Windows Azureソリューションの使用に関連する運用コストを削減できます。

キューベースのメッセージングソリューションを実装する場合、開発者はストレージとのデータ交換トランザクションの数を減らすことができます。

1. メッセージをキューに送信する場合、相互に関連するメッセージを1つの大きなパッケージにグループ化し、圧縮イメージを圧縮してBLOBストレージに保存し、キューを使用してこのデータを含むBLOBへのリンクを保存できます。
2. キューからメッセージを受信する場合、複数のメッセージを1つのパッケージに結合して、リポジトリとのデータ交換でトランザクションを実行できます。 キューサービスAPIに実装されているGetMessagesメソッドは、指定された数のメッセージが単一のトランザクション内のキューから削除されるようにします（以下の注を参照）。
3. キュー内の作業項目の存在を確認するときは、 積極的なポーリング間隔の使用を避け 、キューへの要求がデータを返さない場合、キューのポーリング間隔を増やす時間遅延を設定します。
4. キューリスナーの数を減らす -クエリベースのモデルを使用する場合、キューが空の場合、各ロールインスタンスに対して1つのリスナーのみを使用します。 各ロールインスタンスのリスナーの数を無効にするには、キューが作業項目を受け取ったときに通知メカニズムを使用してリスナーインスタンスを作成します。
5. ほとんどの場合、ワークキューが空のままの場合 、システムメトリックを監視するロールインスタンスの数を自動的に削減するメカニズムを作成して 、増加したワークロードを処理するためにアプリケーションがロールインスタンスの数を増やす必要がある時期を判断します。

上記の推奨事項は、メッセージパケットを処理し、キュー、BLOBストレージ、およびフロー制御と対話する基本的な操作のほとんどをカプセル化するように設計された一般的なメカニズムとして実装できます。 次に、このようなメカニズムを実装する方法を検討します。

重要な情報 。 GetMessagesメソッドを使用してメッセージを受信する場合、キューからの削除操作のキューサービスAPIパッケージの最大サイズは32です。この値を超えると、ランタイム例外が発生します。

Windows Azureキューのトランザクションコストは、キューサービスクライアントの数が増えると、たとえばロールインスタンスの数を増やしたり、削除スレッドの数が増えたりすると、直線的に増加します。 上記の推奨事項を考慮せずにソリューションを実装する場合のコストの増加の可能性を示すために、具体的な数値を使用した例を示します。

コストに対する非効率的なアーキテクチャの影響

最適化メカニズムを含まない上記の課金システムのアーキテクチャを作成すると、Windows Azureプラットフォームにソリューションを展開した後に運用コストが増加します。 このセクションでは、追加費用が発生する理由について説明します。

シナリオの定義に従って、ソフトウェアソリューションは定期的にビジネストランザクションデータを受信します。 このソリューションが、標準的な8時間労働時間のわずか25％の作業負荷の処理で忙しいとします。 その結果、システムがトランザクション処理に関与していない場合、6時間（8時間* 0.75）は「非アクティブ時間」になります。 さらに、ソリューションは毎日16時間、データをまったく受信しません。

合計22時間の非アクティブ期間中、ソリューションは新しいデータの追加に関する通知を受信せずに作業情報をキューから削除しようとします。 この間、キューからの個々の削除スレッドは、入力キューに関連付けられた最大79,200トランザクション（22時間* 60分* 60トランザクション/分）を実行し、デフォルトのポーリング間隔は1秒です。

前述のように、Windows Azure Platformの価格モデルでは、ベースユニットとして個別の「ストレージトランザクション」が使用されます。 リポジトリトランザクションは、リポジトリデータを追加、読み取り、更新、または削除するためのユーザーアプリケーションの要求です。 このホワイトペーパーの執筆時点では、倉庫のトランザクションコストは10,000トランザクションあたり0.01ドルでした。 更新 ：転送の公開時： 100,000トランザクションあたり0.01ドル。

重要な情報 。 キューに関連付けられているトランザクションの数を計算する場合、1つのメッセージをキューに入れることは1つのトランザクションであることに注意する必要がありますが、メッセージを受信することは、メッセージの受信とキューからのメッセージの削除要求を含む2ステップのプロセスであることがよくあります。 その結果、キューからメッセージを削除する操作を成功させるには、2つのストレージトランザクションが必要になります。 注：キューからメッセージを削除するリクエストがデータの受信に関連していない場合でも、それは有料トランザクションと見なされます。

上記のシナリオで単一のキュー削除フローによって作成されたVaultトランザクションは、月額サービス請求書に約2.38ドル（$ 79,200 / 10,000 * $ 0.01 * 30日）を追加します。 キューからメッセージを削除する200スレッド（または作業ロールの200コピーに1スレッド）を使用すると、毎月の費用が457.2ドル増加します（ 更新 ：記事の翻訳の公開時に計算を行う場合、これは45.7ドルです）。 これらのコストは、システムが計算を実行せず、キュー内の作業項目の存在のみをチェックする場合に発生します。 上記の例は抽象的です。誰もこの方法でサービスを実装しないからです。 次の最適化手法を使用する必要があります。

不要な遅延を排除するための推奨事項

Windows Azureキューベースのメッセージングシステムのパフォーマンスを最適化するには、以下で説明するように、Windows Azure Integration Busが提供するパブリッシャーおよびサブスクライバーのメッセージ処理レイヤーを使用できます。
この場合、開発者はポーリングと強制通知メカニズムをリアルタイムで組み合わせて、学生が特定の条件下で発生し、新しいワークロードがキューに入れられていることを示す通知イベント（トリガー）にサブスクライブできるようにする必要があります。 このアプローチにより、パブリッシャーおよびサブスクライバーレベルで標準のキューポーリングサイクルを作成して、通知をディスパッチできます。

複雑な分散システムでは、このアプローチでは、「メッセージバス」または「メッセージ処理ミドルウェア」を使用して、1人以上のサブスクライバーに通知を確実に送信する必要があります。 Windows Azure Integration Busは、Windows Azureだけでなくローカルに展開された疎結合分散アプリケーションサービス間の最適なメッセージングソリューションです。 キューを使用してデータ転送プロセス間で通知を交換できる「メッセージバス」アーキテクチャの実装に最適です。

キューを使用してメッセージングシステムを作成する手順では、次のパターンを使用できます。

パブリッシャーとキューサービスサブスクライバー間のやり取りでWindows Azureロールのインスタンス間でデータを交換するために使用される原則は、ほとんどのプッシュ通知交換要件を満たしています。 このプロセスの基本概念は、以前の出版物のいずれかで説明されています。

重要な情報 。 Windows Azure統合バスの使用は、このプロセスの2つの重要な要素を考慮した価格体系によって規制されています。 まず、データセンターの交換時にデータの送受信に料金がかかります。 第二に、アプリケーションと統合バスインフラストラクチャ間に確立された接続の数に対して料金が課金されます。
この点で、統合バスを使用して特定のアーキテクチャを実装することのすべてのプラス面とマイナス面を評価するために、コストと利点を分析することが重要です。 統合バスに基づく通知ディスパッチレベルの実装が実際にコストを削減するかどうかを評価する必要があります。これにより、このプロジェクトへの投資と開発者の追加人件費を正当化できます。

パブリッシャーとサブスクライバーの間に追加のレベルのメッセージングを作成することにより、遅延の悪影響を非常に簡単に最小限に抑えることができます。 追加のコスト削減は、動的（弾性）スケーリングによって実現されます。その実装については、次のセクションで説明します。

動的スケーリングのガイドライン

Windows Azureプラットフォームは、顧客ソリューションを簡単かつ迅速にスケールアップおよびスケールダウンする機能をサポートしています。 ワークロードとトラフィックの変動に適応できることは、このクラウドコンピューティングプラットフォームの主な利点の1つです。 これは、「スケーラビリティ」の概念がIT専門家の辞書の用語ではなくなったことを意味し、スケーラビリティのサポートに過剰なコストは不要になりました。 この機能のソフトウェア実装は、十分に開発されたアーキテクチャを備えたクラウドソリューションで利用できます。

動的スケーリングは特定のソリューションの技術的特徴であり、ランタイムで使用可能なストレージスペースとコンピューティングパワーを増減することにより、さまざまなワークロードに適応できます。 Windows Azure Platformには、消費者が設定料金で必要な電力を割り当てることができる分散コンピューティングインフラストラクチャを使用した動的スケーリングのサポートが組み込まれています。

Windows Azureプラットフォームでサポートされている2種類の動的スケーリングを区別することが重要です。

• ロールインスタンスのスケーリングとは、現在のワークロードを処理するための作業ロールまたはWebロールの追加インスタンスの追加と削除です。 これには、多くの場合、サービス構成内のインスタンスの数を変更する必要があります。 Windows Azureランタイムは、インスタンスの数の増加に対応して新しいインスタンスを作成し、インスタンスの数を減らすと、既存のインスタンスの一部がシャットダウンします。
• プロセス （スレッド）のスケーリングとは 、現在のワークロードに応じてスレッドの数を増減することにより、ロールの特定のインスタンスに対して十分なレベルの容量（処理スレッドの数）を維持することです。

役割ベースのメッセージングソリューションに動的スケーリングを実装するには、次の推奨事項が必要です。

1. CPU使用率、キューの深さ、応答時間、メッセージ処理遅延など、 主要なパフォーマンスメトリックを追跡します 。
2. 作業ロールインスタンスの数を動的に増減して、予想されるものと予測できないものの両方のピークワークロードを処理します 。
3. プログラムで処理スレッドの数を増減して 、ワークロードのさまざまなインジケーターにシステムを適合させます。
4. .NET Framework 4のTask Parallel Libraryを使用した、 ワークロードのパーティション分割と小さなフラグメントの並列処理 。
5. さまざまなレベルのワークロードでソリューションを管理する際に、コンピューティングパワーの可用性を確保します 。 これにより、追加のコピーを作成するための追加の努力をすることなく、負荷の突然の増加に対処することができます。

サービス管理APIを使用すると、Windows Azureプラットフォームでホストされるサービスは、実行時に展開構成を変更することにより、実行中のロールインスタンスの数を増減できます。

ご注意 既定では、標準のサブスクリプションで最大20のWindows Azureコンピューティング操作のインスタンスを使用できます。 これにより、Windows Azureプラットフォームのユーザーが誤って非常に多くのロールインスタンスを作成しようとした場合に、サービスコストが増加するのを防ぎます。 これは、いわゆる「ソフト」制限です。 このクォータを増やす要求は、Windows Azureサポートテクニカルサポートチームに送信する必要があります。

ロールインスタンスの数を動的にスケーリングすることが、劇的に増加したワークロードを処理するための最良の方法であるとは限りません。 たとえば、仮想マシンの新しいインスタンスは、作業の準備が整うまでに数秒かかります。現在、サービスレベル契約では、このプロセスの期間に関連するインジケーターは提供されていません。 代わりに、より簡単な方法で、一時的なワークロードの増加に対処するためにワークフローの数を増やすことができます。 ワークロードを処理するとき、そのインジケータを監視して、ワークプロセスの数の動的な増加または減少を必要とする状況を特定します。

重要な情報 。 現在、単一のWindows Azureキューのスケーラビリティメトリックターゲットは、1秒あたり500トランザクションに制限されています。 アプリケーションがこのしきい値を超えようとすると、たとえば、キュ​​ーからオブジェクトを削除する数百のスレッドが実行されている複数のロールインスタンスを使用してキューで操作を実行すると、ストレージサービスはHTTPエラー503 "サーバーがビジーです"を返すことがあります このエラーが発生した場合、アプリケーションは、遅延時間が指数関数的に増加するアルゴリズムを使用して、トランザクションの再試行メカニズムを実装する必要があります。 ただし、HTTP 503エラーが定期的に発生する場合は、複数のキューを使用し、これらのキューを使用してワークロードをスケーリングできるセグメンテーション戦略を適用することをお勧めします。

ほとんどの場合、ワークフローの自動スケーリングは、ロールの個別のインスタンスによって実行されます。 役割インスタンスのスケーリングには、多くの場合、パフォーマンスメトリックを追跡し、システムをスケーリングするための手順を実行するソリューションアーキテクチャの中心的な要素の開発が必要です。 次の図は、 ダイナミックスケーリングエージェントと呼ばれるサービスコンポーネントを示しています。これは、ワークロードメトリックに関連するデータを収集および分析して、新しいインスタンスまたは使用停止が非アクティブかどうかを判断します。

Scaling Agentサービスは、Windows Azureプラットフォームの作業ロールとして、またはローカルサービスとして展開できます。 使用される展開トポロジに関係なく、このサービスはWindows Azureキューにアクセスできます。

遅延時間のスケーリングへの影響、データをストレージと交換する際のトランザクションコスト、および動的スケーリングの要件について説明したので、これらの推奨事項の実際の実装に移りましょう。

技術的な実装

前のセクションでは、Windows Azureキューストレージサービスを使用して実装された、適切に設計されたメッセージングアーキテクチャの主要な機能について説明しました。 スケーリングの3つの重要な側面を検討しました。データ処理の遅延の削減、ストレージトランザクションコストの最適化、ワークロードの不安定性への応答性の改善です。

このセクションは、Windows Azureアプリケーションの開発者を対象としており、テンプレートのソフトウェア実装について説明しています。

ご注意 このセクションには、クエリベースのモデルとプッシュデータだけでなく、自動スケーリングをサポートするキューリスナーの作成に関する情報が含まれています。 ロールインスタンスレベルでの動的なスケーリングの最新の方法については、ユーザーコミュニティによって実装され、 MSDN Code Gallery Webサイトで公開されているプロジェクトを参照してください。

標準キューリスナーの作成

まず、キューリスナーコンポーネントによって実装されるコントラクトを作成します。これは、作業ロールに配置され、Windows Azureキューのデータ転送を想定しています。

///    ,     Windows Azure. public interface ICloudQueueServiceWorkerRoleExtension { ///    ,         . void StartListener(int threadCount); ///            . CloudQueueListenerInfo QueryState(); ///             Windows Azure. int DequeueBatchSize { get; set; } ///      ,           . TimeSpan DequeueInterval { get; set; } ///    , ,   . event WorkCompletedDelegate QueueEmpty; } 

QueueEmptyイベントは、ノードによって使用されます。 キューが空のときにノードがキューリスナーの動作を制御できるメカニズムが含まれています。 イベントデリゲートは次のように定義されます。

 /// <summary> ///    ,        ///       . /// </summary> /// <param name="sender"> .</param> /// <param name="idleCount"> ,    .</param> /// <param name="delay">,             .</param> /// <returns>,              .</returns> public delegate bool WorkCompletedDelegate(object sender, int idleCount, out TimeSpan delay); 

CloudQueueMessageなどのSDKに組み込まれたクラスを使用する代わりに、汎用テンプレートをサポートするリスナーを作成することにより、キュー要素の処理を簡素化できます。 ユニバーサルキューアクセスパターンをサポートする新しいキューリスナーインターフェイスを作成します。

 /// <summary> ///     ,       . /// </summary> /// <typeparam name="T">   ,     .</typeparam> public interface ICloudQueueListenerExtension<T> : ICloudQueueServiceWorkerRoleExtension, IObservable<T> { } 

注：また、.NET Framework 4で利用可能なIObservableインターフェイスを使用してObserverデザインテンプレートを実装することにより、1つまたは複数のサブスクライバーにキュー要素を送信するユニバーサルテンプレートをサポートするリスナーを許可しました。
ICloudQueueListenerExtensionインターフェイスが実装されているコンポーネントの1つのインスタンスを保存する予定です。 ただし、複数のスレッド（タスク）を同時に実行してキューからメッセージを削除する機能が必要です。 したがって、リスナコンポーネントのキューからメッセージを削除するためのマルチスレッドロジックのサポートを追加します。 この問題を解決するために、並列データ処理タスク並列ライブラリ（TPL）の機能のライブラリが使用されます。 StartListenerメソッドを使用すると、キューからメッセージを削除するために必要な数のスレッドを作成できます。

 /// <summary> ///        . /// </summary> /// <param name="threadCount">     .</param> public void StartListener(int threadCount) { Guard.ArgumentNotZeroOrNegativeValue(threadCount, "threadCount"); //             . if (this.dequeueTasks.IsAddingCompleted) { this.dequeueTasks = new BlockingCollection<Task>(this.dequeueTaskList); } for (int i = 0; i < threadCount; i++) { CancellationToken cancellationToken = this.cancellationSignal.Token; CloudQueueListenerDequeueTaskState<T> workerState = new CloudQueueListenerDequeueTaskState<T>(Subscriptions, cancellationToken, this.queueLocation, this.queueStorage); //             ,      . this.dequeueTasks.Add(Task.Factory.StartNew(DequeueTaskMain, workerState, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)); } //    ,    . this.dequeueTasks.CompleteAdding(); } 

DequeueTaskMainメソッドは、キューからメッセージを削除するスレッドの機能を実装します。 次の基本操作をサポートしています。

 /// <summary> ///        Windows Azure. /// </summary> /// <param name="state">,    .</param> private void DequeueTaskMain(object state) { CloudQueueListenerDequeueTaskState<T> workerState = (CloudQueueListenerDequeueTaskState<T>)state; int idleStateCount = 0; TimeSpan sleepInterval = DequeueInterval; try { //          ,             . while (workerState.CanRun) { try { var queueMessages = from msg in workerState.QueueStorage.Get<T>(workerState.QueueLocation.QueueName, DequeueBatchSize, workerState.QueueLocation.VisibilityTimeout).AsParallel() where msg != null select msg; int messageCount = 0; //            PLINQ. queueMessages.ForAll((message) => { //     . idleStateCount = 0; //       ,   . workerState.OnNext(message); //    ,     . workerState.QueueStorage.Delete<T>(message); //       . messageCount++; }); // ,       . if (0 == messageCount) { //     ,        (,     ). idleStateCount++; //    ,      . if (QueueEmpty != null) { // ,          . if (QueueEmpty(this, idleStateCount, out sleepInterval)) { //      ,      . break; } } //        . Thread.Sleep(sleepInterval); } } catch (Exception ex) { if (ex is OperationCanceledException) { throw; } else { //           . workerState.OnError(ex); //        ,         . Thread.Sleep(sleepInterval); } } } } finally { workerState.OnCompleted(); } } 

DequeueTaskMainメソッドの実装機能に関連するいくつかの説明を行う必要があります。
最初に、Parallel LINQ（PLINQ）メソッドを使用して、後続の処理のためにメッセージをディスパッチします。

この問題を解決するためにPLINQを使用する主な利点は、可能な場合はいつでも異なるプロセッサ上の個別のワークフローでデリゲートを並行して使用するため、メッセージ処理が高速化されることです。

ご注意 内部クエリの並列化制御は、PLINQによって提供されます。 PLINQシステムが並列化をサポートするために複数のコアを使用するという保証はありません。 PLINQシステムが、並列化のための計算能力の追加コストのためにクエリの実行が遅くなる可能性を検出した場合、クエリを順次実行できます。 PLINQのすべての利点を実現するには、クエリの総ワークロードが、スレッドプールを管理するための追加の計算能力の使用を正当化するのに十分な大きさである必要があります。

第二に、特定のメッセージを受信するための個別のリクエストを作成しません。 代わりに、キューサービスAPIを使用して、キューから指定された数のメッセージを取得します。 受信されるメッセージの数は、Getメソッドに渡されるDequeueBatchSizeパラメーターによって決まります。 データウェアハウスの抽象化レベルにアクセスすると、このパラメーターはキューサービスAPIに渡されます。 さらに、パケットサイズがAPIで許可されている最大値を超えないようにするために、セキュリティチェックが実行されます。 以下は、このアプローチのソフトウェア実装です。

 ///          Windows Azure    . public sealed class ReliableCloudQueueStorage : ICloudQueueStorage { ///   ,   API  Queue Service     Get. private const int MaxDequeueMessageCount = 32; ///             . public IEnumerable<T> Get<T>(string queueName, int count, TimeSpan visibilityTimeout) { Guard.ArgumentNotNullOrEmptyString(queueName, "queueName"); Guard.ArgumentNotZeroOrNegativeValue(count, "count"); try { var queue = this.queueStorage.GetQueueReference(CloudUtility.GetSafeContainerName(queueName)); IEnumerable<CloudQueueMessage> queueMessages = this.retryPolicy.ExecuteAction<IEnumerable<CloudQueueMessage>>(() => { return queue.GetMessages(Math.Min(count, MaxDequeueMessageCount), visibilityTimeout); }); // ...     ... 

結論として、キューからメッセージを無期限に削除するタスクを完了するつもりはないことに注意してください。 キューが空になるたびに呼び出されるQueueEmptyイベントとして実装された、明示的に定義されたチェックポイントを作成しました。 この場合、 QueueEmptyイベントハンドラーが呼び出され、キューからメッセージを削除するタスクを完了できるかどうかが決定されます。 QueueEmptyイベントハンドラーの正しい実装は、次のセクションで説明する「自動縮小」のサポートを提供します。

キューからメッセージを削除するタスクを自動的に削減します

QueueEmptyイベントハンドラーを使用すると、2つのカテゴリのタスクを解決できます 。 まず、キューからメッセージを削除する元のタスクにメッセージを送信し、指定された時間間隔（イベントデリゲート遅延の出力パラメーターで指定）でスリープモードに入るコマンドを送信します。 次に、送信された論理パラメーターを使用して、作業を完了する必要性についてメッセージをキューから削除するようタスクに通知します。

QueueEmptyイベントハンドラーの次の実装により、上記の両方の問題を解決できます。 ハンドラーは要求間の間隔を計算し、2つの連続したポーリング間の遅延の指数関数的増加の必要性についてメッセージをキューから削除するようタスクに通知します。 注：このソリューションでは、リクエスト間の遅延は1秒を超えません。 自動スケーリングが正しく実装されていれば、ポーリング操作の間に長い遅延は必要ありません。 さらに、イベントハンドラーはキューリスナーの状態を照会して、キューからメッセージを削除するためのアクティブなタスクの数を判断します。 この数が1を超える場合、イベントハンドラーは、要求間の待機間隔の最大値に達したときに、メッセージをキューから削除する最初のタスクがポーリングサイクルを完了することを推奨します。 それ以外の場合、キューからメッセージを削除するタスクは中断されません。これにより、キューリスナーの各インスタンスで一度に実行されている要求ストリームを1つだけ残すことができます。 このアプローチは 、前述のように、 ストレージトランザクションの数を減らし、トランザクションコストを削減するのに役立ちます 。

 private bool HandleQueueEmptyEvent(object sender, int idleCount, out TimeSpan delay) { //     ICloudQueueServiceWorkerRoleExtension,     . ICloudQueueServiceWorkerRoleExtension queueService = sender as ICloudQueueServiceWorkerRoleExtension; //    ,         . IWorkItemProcessorConfigurationExtension config = Extensions.Find<IWorkItemProcessorConfigurationExtension>(); //              . CloudQueueListenerInfo queueServiceState = queueService.QueryState(); //   ,   . int deltaBackoffMs = 100; int minimumIdleIntervalMs = Convert.ToInt32(config.Settings.MinimumIdleInterval.TotalMilliseconds); int maximumIdleIntervalMs = Convert.ToInt32(config.Settings.MaximumIdleInterval.TotalMilliseconds); //               . int delta = (int)((Math.Pow(2.0, (double)idleCount) - 1.0) * (new Random()).Next((int)(deltaBackoffMs * 0.8), (int)(deltaBackoffMs * 1.2))); int interval = Math.Min(minimumIdleIntervalMs + delta, maximumIdleIntervalMs); //                 . delay = TimeSpan.FromMilliseconds((double)interval); //               , //          .    ,        . return delay.TotalMilliseconds >= maximumIdleIntervalMs && queueServiceState.ActiveDequeueTasks > 1; } 

自動スケーリングメカニズムは、次のように説明できます。

• キューにメッセージが表示されるとすぐに、キューからメッセージを削除するタスクがワークロードをすぐに処理します。 キューからメッセージパケットを削除する要求は、遅延なく送信されます。
• 元のキューが空になった後、キューからメッセージを削除する各タスクはQueueEmptyイベントを発生させます。
• QueueEmptyイベントハンドラーは、要求の送信間のランダムな指数関数的な遅延を計算し、キューからメッセージを削除して指定された間隔で操作を一時停止するようタスクに指示します。
• キューからメッセージを削除するタスクは、累積非アクティブ時間が許容最大値を超えるまで、指定されたランダムな間隔で初期キューをポーリングし続けます。
• 非アクティブ間隔の最大長に達し、元のキューがまだ空の場合、キューからメッセージを削除するためのすべてのアクティブなタスクが終了します。 メッセージの送信間隔を計算するアルゴリズムは異なる時間に機能するため、これは同時には発生しません。
• キューからメッセージを削除するタスクが1つだけアクティブのままになる場合があります。 その結果、空のキューのポーリングに関連するトランザクションの処理は、この残りのタスクによって生成されたものを除き、停止されます。

ワークロードの現在のレベルに関する情報を収集するメカニズムの詳細な調査のために、ソースコードの対応するアーティファクトの検討に移ります。 第1に、結果として生じるシステムのワークロードの測定を担当するインジケーターを格納する構造があります。 簡単な例として、サンプルコードで使用するインジケーターの小さなサブセットを示します。

 ///  ,        . public struct CloudQueueListenerInfo { ///       Windows Azure. public int CurrentQueueDepth { get; internal set; } ///       ,       . public int ActiveDequeueTasks { get; internal set; } ///         ,  . public int TotalDequeueTasks { get; internal set; } } 

次に、キューリスナーは、その負荷メトリックを返すメソッドを使用します（以下の例を参照）。

 ///            . public CloudQueueListenerInfo QueryState() { return new CloudQueueListenerInfo() { CurrentQueueDepth = this.queueStorage.GetCount(this.queueLocation.QueueName), ActiveDequeueTasks = (from task in this.dequeueTasks where task.Status != TaskStatus.Canceled && task.Status != TaskStatus.Faulted && task.Status != TaskStatus.RanToCompletion select task).Count(), TotalDequeueTasks = this.dequeueTasks.Count }; } 

キュー削除タスクの自動スケーリング

前のセクションでは、アクティブな削除タスクの数をキューから1つのインスタンスに減らして、ストレージとのデータ交換操作に関連するコストの増加に対する非アクティブモードトランザクションの影響を減らす機能について説明しました。 このセクションでは、計算能力を向上させる自動スケーリング機能の実装例を示します。

自動スケーリングなどのイベントをトリガーするために、空のキューから空でないキューへの遷移を追跡するデリゲートを作成します。

 /// <summary> ///  ,        ,    . /// </summary> /// <param name="sender"> .</param> public delegate void WorkDetectedDelegate(object sender);     ICloudQueueServiceWorkerRoleExtension,      ,   ,        (       ): public interface ICloudQueueServiceWorkerRoleExtension { // ...    ,   . .        ... ///  ,        ,    . event WorkDetectedDelegate QueueWorkDetected; } 

キューリスナーのプログラムコードのどの行でイベントをトリガーするかを決定しましょう。 この場合、 QueueWorkDetectedイベントは、 DequeueTaskMainメソッドを使用して実装されたデキューループから呼び出されます。このメソッドは、次のように展開する必要があります。

 public class CloudQueueListenerExtension<T> : ICloudQueueListenerExtension<T> { //  ,        ,    . public event WorkDetectedDelegate QueueWorkDetected; private void DequeueTaskMain(object state) { CloudQueueListenerDequeueTaskState<T> workerState = (CloudQueueListenerDequeueTaskState<T>)state; int idleStateCount = 0; TimeSpan sleepInterval = DequeueInterval; try { //          ,             . while (workerState.CanRun) { try { var queueMessages = from msg in workerState.QueueStorage.Get<T>(workerState.QueueLocation.QueueName, DequeueBatchSize, workerState.QueueLocation.VisibilityTimeout).AsParallel() where msg != null select msg; int messageCount = 0; //      ,   . if (idleStateCount > 0 && queueMessages.Count() > 0) { if (QueueWorkDetected != null) { QueueWorkDetected(this); } } // ...    ,   . .        ... 

最後の手順で、 QueueWorkDetectedイベントハンドラーを作成します。 このイベントハンドラーの実装は、キューリスナーをインスタンス化するコンポーネントに含まれます。 私たちの場合、このコンポーネントは作業ロールです。 イベントハンドラの実装とそのインスタンスの作成を担当するプログラムコードは、次の部分で構成されています。

 public class WorkItemProcessorWorkerRole : RoleEntryPoint { //  Windows Azure    . public override sealed bool OnStart() { // ...      ... //       . var inputQueueListener = new CloudQueueListenerExtension<XDocument>(inputQueueLocation); //    . inputQueueListener.QueueEmpty += HandleQueueEmptyEvent; inputQueueListener.QueueWorkDetected += HandleQueueWorkDetectedEvent; inputQueueListener.DequeueBatchSize = configSettingsExtension.Settings.DequeueBatchSize; inputQueueListener.DequeueInterval = configSettingsExtension.Settings.MinimumIdleInterval; // ...     ... } ///  ,        ,    . private void HandleQueueWorkDetectedEvent(object sender) { //     ICloudQueueServiceWorkerRoleExtension,     . ICloudQueueServiceWorkerRoleExtension queueService = sender as ICloudQueueServiceWorkerRoleExtension; //              . CloudQueueListenerInfo queueServiceState = queueService.QueryState(); //         ,     . int dequeueTaskCount = GetOptimalDequeueTaskCount(queueServiceState.CurrentQueueDepth); //             ,         . if (queueServiceState.ActiveDequeueTasks < dequeueTaskCount) { //       . queueService.StartListener(dequeueTaskCount - queueServiceState.ActiveDequeueTasks); } } // ...     ... 

上記の例に照らして、GetOptimalDequeueTaskCountメソッドの使用を明確にする必要があります。このメソッドは、ワークロードを処理するキュー削除タスクの数を計算します。呼び出すとき、キューリスナーが予想されるワークロードを処理するために必要な計算能力を（適切な意思決定メカニズムを使用して）決定する必要があります。

たとえば、開発者は最も単純なパスに従って、GetOptimalDequeueTaskCountメソッドで静的ルールのセットを直接実装できます。。キュー処理インフラストラクチャのスループットとスケーラビリティの既知のパラメーター、平均遅延時間、処理されたデータの量、およびその他の情報に基づいて、このルールのセットは、キューから削除し、それらを増やすことを決定する必要なタスク数の楽観的な推定値を与えることができます。

次の例では、単純化された方法を使用して、キューからの削除タスクの最適な数を決定します。

 /// <summary> ///         ,     . /// </summary> /// <param name="currentDepth">    .</param> /// <returns>     .</returns> private int GetOptimalDequeueTaskCount(int currentDepth) { if (currentDepth < 100) return 10; if (currentDepth >= 100 && currentDepth < 1000) return 50; if (currentDepth >= 1000) return 100; //    . return 1; } 

上記のコード例は、普遍的な解決策にはなりません。理想に近いソリューションは、外部から構成と制御をサポートし、必要な計算をすべて実行できるルールを呼び出すことです。

現在、ワークロードの変動（増減）に応じて自動的にスケーリングできるキューリスナーの作業プロトタイプがあります。処理中のワークロードの変動に適応するために、おそらくその機能を拡張する必要があります。QueueWorkDetectedイベントのサポートを追加するために使用したものと同じテンプレートを適用することにより、この関数を追加できます。

次に、キューリスナーを操作する際の遅延時間を短縮するための別の重要な最適化方法を検討します。

遅延なしでキューから削除するパブリッシャーおよびサブスクライバーレベルの実装

このセクションでは、統合バスの単方向マルチキャストに基づくプッシュ通知メカニズムを使用して、キューリスナーの上記の実装を補完します。この通知メカニズムはトリガーイベントを処理します。トリガーイベントは、キューからアイテムを削除する作業を開始する必要性についてのリスナーへの信号です。このアプローチにより、新しいメッセージをチェックするときにキューのポーリングを拒否でき、遅延要因がなくなります。

新しいワークロードがキューリスナーに表示されたときにキューリスナーが受け取るトリガーイベントを作成します。

 ///  -,         . [DataContract(Namespace = WellKnownNamespace.DataContracts.Infrastructure)] public class CloudQueueWorkDetectedTriggerEvent { ///     ,    . [DataMember] public string StorageAccount { get; private set; } ///   ,     . [DataMember] public string QueueName { get; private set; } ///      (,       ). [DataMember] public long PayloadSize { get; private set; } // ...      ... } 

キューリスナーの実装をトリガーイベントのサブスクライバーとして機能させます。最初の手順は、キューリスナーをCloudQueueWorkDetectedTriggerEventイベントのオブザーバーとして指定することです。

 ///    ,     Windows Azure. public interface ICloudQueueServiceWorkerRoleExtension : IObserver<CloudQueueWorkDetectedTriggerEvent> { // ...  , . .      ... } 

第二段階は、メソッドの実装になりOnNext IObserverインターフェースで定義されました、。このメソッドは、新しいイベントをオブザーバーに通知するためにプロバイダーによって呼び出されます。

 public class CloudQueueListenerExtension<T> : ICloudQueueListenerExtension<T> { // ...      ... /// <summary> ///          /// </summary> /// <param name="e">-,        .</param> public void OnNext(CloudQueueWorkDetectedTriggerEvent e) { Guard.ArgumentNotNull(e, "e"); // ,  -   ,  ;  ,    . if (this.queueLocation.StorageAccount == e.StorageAccount && this.queueLocation.QueueName == e.QueueName) { if (QueueWorkDetected != null) { QueueWorkDetected(this); } } } // ...     ... } 

前の例からわかるように、前の手順で使用したのと同じデリゲートを意図的に使用します。QueueWorkDetectedイベントハンドラーは、最適な数のキュー削除タスクインスタンスを作成するアプリケーションロジックを提供します。したがって、CloudQueueWorkDetectedTriggerEvent通知を処理するときに同じイベントハンドラーを再利用します。

前のセクションで説明したように、プッシュ通知を使用する場合、常に実行されているキュー削除タスクを維持する必要はありません。これにより、キューリスナーの各インスタンスで発生するキュー処理タスクの数をゼロに減らすことができます。また、通知メカニズムを使用して、キューが作業項目を受け取ったときにキューから削除タスクのインスタンスを作成できます。キューから非アクティブな削除タスクがないことを確認するには、QueueEmptyイベントハンドラーに簡単な変更を加えます。

 private bool HandleQueueEmptyEvent(object sender, int idleCount, out TimeSpan delay) { // ...      ... //               . return delay.TotalMilliseconds >= maximumIdleIntervalMs; } 

したがって、キューからアクティブな削除タスクが1つあるかどうかは判断できなくなりました。QueueEmpty変更イベントハンドラーは、最大非アクティブ間隔が超過したという事実のみを考慮し、その後、キューからのすべてのアクティブな削除タスクが完了します。CloudQueueWorkDetectedTriggerEvent

通知を受信するには、パブリッシャーおよびサブスクライバーモデルを使用します。これは、Windows Azureロールのインスタンス間で疎結合メッセージングモデルとして実装されます。本質的に、ロール間のデータ交換のレベルを使用して、次のように着信イベントを処理します。

 public class InterRoleEventSubscriberExtension : IInterRoleEventSubscriberExtension { // ...      .     . ,     Windows Azure Customer Advisory Team ... public void OnNext(InterRoleCommunicationEvent e) { if (this.owner != null && e.Payload != null) { // ...      ... if (e.Payload is CloudQueueWorkDetectedTriggerEvent) { HandleQueueWorkDetectedTriggerEvent(e.Payload as CloudQueueWorkDetectedTriggerEvent); return; } // ...     ... } } private void HandleQueueWorkDetectedTriggerEvent(CloudQueueWorkDetectedTriggerEvent e) { Guard.ArgumentNotNull(e, "e"); //        -. foreach (var queueService in this.owner.Extensions.FindAll<ICloudQueueServiceWorkerRoleExtension>()) { //  -  . queueService.OnNext(e); } } } 

CloudQueueWorkDetectedTriggerEventクラスで定義されたトリガーイベントマルチキャストは、パブリッシャー、つまり作業項目をキューに入れるコンポーネントの責任です。このイベントは、最初の作業項目がキューに送信される前、または最後の作業項目がキューに配置された後に発生できます。次の例では、すべての作業項目を入力キューに配置した後、トリガーイベントを発行します。

 public class ProcessInitiatorWorkerRole : RoleEntryPoint { //   ,       . private volatile IInterRoleCommunicationExtension interRoleCommunicator; // ...      .     . ,     Windows Azure Customer Advisory Team ... private void HandleWorkload() { //  1.   ,     . // ... (    ) ... //  2.      . // ... (    ) ... //  3.       . //  -,   ,        . var trigger = new CloudQueueWorkDetectedTriggerEvent("MyStorageAccount", "InputQueue"); //       . var interRoleEvent = new InterRoleCommunicationEvent(CloudEnvironment.CurrentRoleInstanceId, trigger); //             . interRoleCommunicator.Publish(interRoleEvent); } } 

そこで、マルチスレッド処理、自動スケーリング、プッシュ通知をサポートするキューリスナーを作成しました。Windows Azureプラットフォーム向けのキューベースのメッセージングソリューションの設計に関するすべての推奨事項を組み合わせるときが来ました。

おわりに

Windows Azureキューベースのメッセージングソリューションの効率とコスト効果を最大化するには、アーキテクトと開発者はこれらのガイドラインに従う必要があります。

ソリューションアーキテクトは以下を考慮する必要があります。

• キューベースのメッセージングアーキテクチャを作成するときは、クラウドソリューションとハイブリッドソリューションのレイヤーとサービス間の非同期でスケーラブルなメッセージングにWindows Azureキューストレージサービスを使用する必要があります。
• 1秒あたり500を超えるトランザクションをサポートするように拡張できるセグメント化されたメッセージングアーキテクチャを使用することをお勧めします。
• Windows Azure ; .
• , .
• .
• 統合バスを使用してリアルタイム通知の準備をする前に、メリットとコストを評価する必要があります。

開発者は次のことを考慮する必要があります。

• メッセージングソリューションの設計では、Windows Azureキューからデータを保存および取得するための手順を作成するときに、バッチ処理を実装する必要があります。
• 効果的なキューリスニングサービスを実装するには、単一のスレッドで空のキューをポーリングして、キューからアイテムを削除するメカニズムを作成する必要があります。
• キューが長時間空のままの場合、動的スケーリングを使用してワーカーロールインスタンスの数を減らします。
• , .
• , , .
• Windows Azure , .
• Windows Azure Integration Bus単方向機能を使用してプッシュ通知をサポートし、待機時間を短縮し、キューベースのメッセージングソリューションのパフォーマンスを向上させます。
• TPL、PLINQ、Observerデザインパターンなど、.NET Framework 4の新機能を活用して、処理の同時実行性を高め、同時実行性を向上させ、マルチスレッドサービスの設計を合理化します。