😍 🤽 👨‍👦 単一のメッセージを見逃さない方法 🚴🏽 👩🏽‍🎓 🌈

イベント処理は、サーバーレステクノロジーで最も一般的なタスクの1つです。今日は、損失を無効にする信頼できるメッセージハンドラを作成する方法について説明します。ちなみに、例はPollyライブラリを使用してC＃で記述されていますが、示されているアプローチは、特に指定のない限り、どの言語でも機能します。

著者に床を渡します。

Azure Functionsによる堅牢なイベント処理

数週間前、関数を使用してイベントを順番に処理する方法に関する記事を公開しました。本日の出版物では、損失を無効にする信頼できるメッセージハンドラを作成する方法の概要を説明します。この記事は2つまたは3つの部分に分けることができますが、すべての情報を1つの資料にまとめることにしました。素晴らしい結果が得られましたが、回路ブレーカーや例外フィルターのパターンを使用するなど、単純なものから最も複雑なものまで、幅広いタスクをカバーしています。例はC＃で記述されていますが、示されているアプローチはどの言語でも機能します（特に指定がない限り）。

分散システムのイベントフローに関連する問題

一定の速度（たとえば、1秒あたり100個）でイベントを送信するシステムを想像してください。 Azure Functionsでこれらのイベントの受信を構成するのは非常に簡単です。わずか数分で、これらの100個のイベントを毎秒処理する多数の同時インスタンスを準備できます。しかし、パブリッシャーが誤って生成されたイベントを送信した場合はどうなりますか？インスタンスの1つが障害のために動作を停止した場合はどうなりますか？または、さらに処理ステップを実行するシステムの1つがシャットダウンしますか？アプリケーションの全体的な整合性とスループットを維持しながら、そのような状況に対処する方法は？

キューを使用するときのメッセージ処理の信頼性を確保することは、わずかに簡単です。 Azure Functionsでは、キューからのメッセージを処理するときに、関数はそのようなメッセージを「ブロック」して処理を試み、失敗した場合はロックを解除して別のインスタンスが受け入れて再試行できるようにします。これらの試行は、メッセージが正常に処理されるか、試行の最大許容回数（デフォルトは4）に達するまで続けられます。 2番目の場合、メッセージは疑わしいメッセージのキューに追加されます。キューからのメッセージがこの試行サイクルを通過しても、キューからの他のメッセージの並列抽出は停止しません。したがって、1つのエラーは全体的なスループットにほとんど影響しません。ただし、ストレージキューは順序を保証せず、サービス（イベントハブなど）に高いスループットを提供するように最適化されていません。

イベントストリーム（Azure Event Hubsなど）はロックを使用しません。これらのサービスは、高帯域幅を提供し、複数の消費者グループとプレイアビリティをサポートするように設計されています。イベントを受信すると、イベントはテープドライブのように機能します。ストリーム内の各セクションには、1つのオフセットポインターがあります。双方向のイベントを読むことができます。イベントストリームの読み取り中にエラーが発生し、ポインターを同じ場所に残すことにしたとします。移動するまで、このセクションのデータをそれ以上処理することはできません。つまり、システムが1秒あたり100個のイベントを引き続き受信するが、Azure Functionsがポインターを新しいイベントに移動することを停止し、誤ったイベントに対処しようとすると、ジャムが発生します。非常に迅速に大量の未処理のイベントを蓄積し、それらは常に成長します。

例外を処理しますが、キューを遅延させません。

オフセットポインターとコンシューマーのこの動作が考慮されました。 関数は、処理が成功したかどうかに関係なく、ポインターを下流に移動します 。これは、システムと関数がそのような状況に対処できる必要があることを意味します。

Azure Functionsがイベントハブからイベントを受け付ける方法

Azure Functionsは、次のようにイベントハブと対話します。

イベントコンセントレーターの各セクションに対して、ポインターが作成され（Azureストレージに配置されます）（ストレージアカウントで確認できます）。
イベントコンセントレータから新しいメッセージを受信すると（デフォルトではバッチモードで実行されます）、ノードはメッセージパケットを渡して機能を開始しようとします。
関数が終了すると（例外の有無に関係なく）、ポインターが移動し、その位置がリポジトリに保存されます。
関数の完了を妨げるものがある場合、ノードはポインターを移動できず、後続のチェックは同じメッセージを（前のコントロールポイントから）受信します。
手順2〜4が繰り返されます。

ここで注意すべきことがいくつかあります。まず、例外を処理しないと、メッセージが失われる可能性があります 。例外が発生して実行が完了しても、ポインターがシフトするためです。 2番目： 関数は、少なくとも1回の配信を保証します （これは、分散システムで一般的な状況です）。これは、同じメッセージが2回受信された状況で、コードとその依存システムが正しく機能することを意味します。以下は、これらの2つの状況の例と、それらに対処するためのコードです。

これらのテストの一環として、シーケンシャル処理用に100,000件のメッセージを公開しました（セクションキーごと）。順序と信頼性を確認して視覚化するために、処理中の各メッセージをRedisキャッシュに記録します。最初のテストでは、100番目のメッセージごとに例外がスローされ、例外処理は実行されません。

[FunctionName("EventHubTrigger")] public static async Task RunAsync([EventHubTrigger("events", Connection = "EventHub")] EventData[] eventDataSet, TraceWriter log) { log.Info($"Triggered batch of size {eventDataSet.Length}"); foreach (var eventData in eventDataSet) { // For every 100th message, throw an exception if (int.Parse((string)eventData.Properties["counter"]) % 100 == 0) { throw new SystemException("Some exception"); } // Insert the current count into Redis await db.ListRightPushAsync("events:" + eventData.Properties["partitionKey"], (string)eventData.Properties["counter"]); } }

このシステムに100,000件のメッセージを送信した後、Redisは次のことを示しました。

ご覧のとおり、100番から112番までの一連のメッセージを見逃しました。どうしたある時点で、関数のインスタンスの1つがこのセクションキーのメッセージパケットを受信しました。この特定のパケットは112番目のメッセージで終了しましたが、セルで例外がスローされました。実行は停止されましたが、機能ノードは引き続き動作し、次のパケットを読み取りました。技術的には、これらのメッセージはイベントコンセントレーターに保存されていましたが、再度処理するには、100番目から112番目までのメッセージを手動で要求する必要があります。

try-catchブロックの追加

この問題を解決する最も簡単な方法は、単にtry / catchブロックをコードに追加することです。これで、例外の場合、ポインターがさらに移動する前に同じプロセスの一部として処理できます。上記のコードにcatchブロックを追加してテストを再開すると、10万件のメッセージがすべて正しい順序で表示されました。

推奨事項：すべてのイベントハブ関数にはcatchブロックが必要です。

この例では、catchブロックを使用してRedisにデータを挿入する追加の試みを行いましたが、通知を送信したり、メッセージを疑わしいキューまたはイベントコンセントレーターに追加処理するなど、他の妥当なアプリケーションを簡単に見つけることができます。

再試行のメカニズムとポリシー

発生する一部の例外は、時々発生する可能性があります。操作を正しく実行するには、単にそれを繰り返すだけで十分な場合があります。前のセクションのコードのcatchブロックでは、1回の再試行が実行されましたが、失敗または例外が発生した場合、メッセージ100〜112が失われます。処理順序を維持しながら、より柔軟な再試行ポリシーを構成できるツールが多数あります。

テストのために、エラー処理にPollyというC＃ライブラリを使用しました。彼女は、単純なリトライポリシーと高度なリトライポリシーの両方を設定できるようにしました。例：「このメッセージを3回挿入してみてください（試行間に遅延がある可能性があります）。すべての試行が失敗した場合は、キューにメッセージを追加してイベントの処理を続行し、後で未処理のメッセージまたは誤ったメッセージに戻るようにします。

 foreach (var eventData in eventDataSet) { var result = await Policy .Handle<Exception>() .RetryAsync(3, onRetryAsync: async (exception, retryCount, context) => { await db.ListRightPushAsync("events:" + context["partitionKey"], (string)context["counter"] + $"CAUGHT{retryCount}"); }) .ExecuteAndCaptureAsync(async () => { if (int.Parse((string)eventData.Properties["counter"]) % 100 == 0) { throw new SystemException("Some Exception"); } await db.ListRightPushAsync("events:" + eventData.Properties["partitionKey"], (string)eventData.Properties["counter"]); }, new Dictionary<string, object>() { { "partitionKey", eventData.Properties["partitionKey"] }, { "counter", eventData.Properties["counter"] } }); if(result.Outcome == OutcomeType.Failure) { await db.ListRightPushAsync("events:" + eventData.Properties["partitionKey"], (string)eventData.Properties["counter"] + "FAILED"); await queue.AddAsync(Encoding.UTF8.GetString(eventData.Body.Array)); await queue.FlushAsync(); } }

このコードでは、エントリを作成するスニペットを使用して、Redisキャッシュにメッセージを追加します。

Redisの最終状態：

より高度な例外キャッチメントポリシーと再試行ポリシーを使用する場合、プリコンパイルされたC＃クラスライブラリには、機能の「例外フィルター」を設定できる機能の試用版が用意されていることに注意してください。これを使用すると、関数の実行中に未処理の例外がスローされたときに実行されるメソッドを作成できます。詳細および例は、この出版物で入手できます。

例外ではないエラーと問題

コードで例外をスローする場合を検討しました。しかし、関数インスタンスがプロセスの中断に遭遇した場合はどうでしょうか？

既に述べたように、Functionが実行を完了しない場合、オフセットポインターはそれ以上移動しません。つまり、メッセージを受信しようとすると、新しいインスタンスは同じデータを受信します。この状況をシミュレートするために、10万のメッセージの処理中に関数アプリケーションを手動で停止、開始、および再起動しました。左側には、結果の一部が表示されます。注：すべてのイベントが処理され、すべてが正常に処理されていますが、一部のメッセージは数回処理されました（700番目以降、601番目以降のメッセージが処理されました）。この動作は少なくとも1回の配信を保証するため、一般にこれは良いことですが、これは私のコードがある程度dem等であるべきであることを意味します。

サーキットブレーカーとコンベア停止

上記のパターンと動作パターンは、再試行を実装するのに便利で、イベントを処理するためのあらゆる努力をするのに役立ちます。特定のレベルの障害は多くの場合許容されます。しかし、多くのエラーが発生し、システムが正常に機能するまで新しいイベントの操作を停止するとします。これは、サーキットブレーカーテンプレートを使用して実現できます。これは、イベント処理チェーンを停止し、後で操作を再開できる要素です。

Polly（再試行を実装したライブラリ）は、いくつかのサーキットブレーカー機能をサポートしています。ただし、これらのパターンは、状態追跡なしでチェーンが複数のインスタンスにまたがる分散時間関数の場合の使用にはあまり適していません。 Pollyを使用してこの問題を解決する方法はいくつかありますが、ここでは必要な機能を手動で追加します。イベント処理でサーキットブレーカーを実装するには、2つのコンポーネントが必要です。

回線の状態を追跡および監視するためのすべてのインスタンスの共通状態。
回路の状態を制御できる（オープンまたはクローズする）メインプロセス。

Redisキャッシュを最初のコンポーネントとして使用し、Azureロジックアプリケーションが2番目になりました。これらの役割は両方とも他の多くのサービスで実行できますが、私はこれら2つのサービスが気に入っています。

すべてのインスタンスの最大許容エラー数

複数のインスタンスがイベントを並行して処理できるため、回線の状態を監視するには、一般的な外部状態が必要です。「30秒以内に、すべてのインスタンスで100を超えるエラーが合計で登録された場合、回線を開いて新しいメッセージの処理を停止する」というルールを実装したかったのです。

Redis TTL追跡機能とソートされたセットを使用して、過去30秒間のエラー数を記録するローリング間隔を取得しました。（詳細に興味がある場合は、これらの例はすべてGitHubで入手できます。）新しいエラーが発生したとき、スライド間隔に目を向けました。エラーの最大数（過去30秒間に100を超える）を超えた場合、イベントをAzure Event Gridサービスに送信しました。対応するRedisコードはこちらから入手できます。そのため、問題を検出し、イベントを送信して回線を開くことができました。

論理アプリケーションを使用した回路状態管理

ステートフルコネクタとオーケストレーションは互いに完全に補完するため、Azureロジックアプリケーションを使用してチェーンの状態を制御しました。回路を開く条件がトリガーされたときに、ワークフロー（Azure Event Gridサービスのトリガー）を開始しました。最初のステップは、Azure Functionsを停止し（Azure Resource Connectorを使用）、通知オプションと応答オプションを含む電子メールを送信することです。その後、すべてが正常に動作していれば、回路の動作を確認して再起動できます。その結果、ワークフローが再開され、機能が起動され、イベントコンセントレーターの最後のコントロールポイントからメッセージ処理が続行されます。

機能を停止した後にロジックアプリケーションから受け取ったメール。 必要に応じて、任意のボタンを押して回路を再開できます。

約15分前に10万件のメッセージを送信し、100件ごとにメッセージがエラーになるようにシステムをセットアップしました。約5,000件のメッセージの後、しきい値を超え、イベントがAzure Event Gridサービスに送信されました。私のAzureロジックアプリケーションはすぐに動作し、機能を停止し、メールを送信しました（上図を参照）。 Redisのコンテンツを見ると、部分的に処理されたセクションが多数表示されます。

リストの一番下は、このセクションキーの最初の200メッセージの処理です。その後、ロジックアプリケーションがシステムを停止しました。

メールのリンクをクリックして、チェーンを再開しました。 Redisで同じクエリを実行すると、イベントコンセントレーターの最後のコントロールポイントから機能が継続して機能していることがわかります。単一のメッセージが失われることはなく、すべてが厳密な順序で処理され、必要なだけ回路を開いたままにすることが判明しました。状態は私のロジックアプリケーションによって制御されていました。

回路を閉じるコマンドの17分前の遅延。

この投稿が、堅牢なメッセージフロー処理のためのAzure Functionsのメソッドとテンプレートの詳細に役立つことを願っています。この知識により、ソリューションの信頼性を損なうことなく、機能を活用することができます（特に、リソースを消費するたびに機能を動的に拡張して料金を支払う）。

リンクをクリックして、この例のさまざまな参照ポイントの各ブランチへのポインターを持つGitHubリポジトリを見つけます。ご質問がある場合は、Twitterで@jeffhollanまでご連絡ください。

単一のメッセージを見逃さない方法