Azure Functions における Azure Event Hubs のトリガーとバインド
この記事では、Azure Functions で Azure Event Hubs のバインドを使用する方法について説明します。 Azure Functions は、イベント ハブのトリガーおよび出力バインドをサポートしています。
| アクション | Type |
|---|---|
| Event Hubs のイベント ストリームに送信されたイベントに応答する | トリガー |
| イベント ストリームにイベントを書き込む | 出力バインド |
拡張機能のインストール
インストールする拡張機能 NuGet パッケージは、関数アプリで使用している C# モードによって異なります。
関数は Functions ホストと同じプロセスで実行されます。 詳細については、「Azure Functions を使用する C# クラス ライブラリ関数を開発する」を参照してください。
拡張機能の機能性は、拡張機能のバージョンによって異なります。
このバージョンでは、シークレットではなく ID を使用して接続する機能が導入されています。 マネージド ID を使用して関数アプリを構成するチュートリアルについては、ID ベースの接続を使用した関数アプリの作成に関 するチュートリアルを参照してください。
このバージョンでは、新しい Event Hubs バインドの種類である Azure.Messaging.EventHubs.EventData が使用されます。
この拡張機能バージョンは、NuGet パッケージ バージョン 5.x をインストールすることによって使用可能です。
バンドルのインストール
Event Hubs 拡張機能は、host.json プロジェクト ファイルで指定される拡張機能バンドルの一部です。 Event Grid バインドのバージョンを変更するために、またはバンドルがまだインストールされていない場合には、このバンドルの変更が必要になることがあります。 詳細については、「拡張機能のバンドル」を参照してください。
このバージョンでは、シークレットではなく ID を使用して接続する機能が導入されています。 マネージド ID を使用して関数アプリを構成するチュートリアルについては、ID ベースの接続を使用した関数アプリの作成に関 するチュートリアルを参照してください。
このバージョンの拡張機能は、host.json ファイルで次のコードを追加するか、または置き換えることによって、プレビュー拡張機能バンドル v3 から追加できます。
{
"version": "2.0",
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[3.3.0, 4.0.0)"
}
}
詳細については、ユーザーの更新に関するページを参照してください。
host.json 設定
host.json ファイルには、Event Hubs トリガーの動作を制御する設定が含まれています。 構成は、拡張機能のバージョンによって異なります。
{
"version": "2.0",
"extensions": {
"eventHubs": {
"maxEventBatchSize" : 10,
"batchCheckpointFrequency" : 1,
"prefetchCount" : 300,
"transportType" : "amqpWebSockets",
"webProxy" : "https://proxyserver:8080",
"customEndpointAddress" : "amqps://company.gateway.local",
"initialOffsetOptions" : {
"type" : "fromStart",
"enqueuedTimeUtc" : ""
},
"clientRetryOptions":{
"mode" : "exponential",
"tryTimeout" : "00:01:00",
"delay" : "00:00:00.80",
"maximumDelay" : "00:01:00",
"maximumRetries" : 3
}
}
}
}
| プロパティ | Default | 説明 |
|---|---|---|
| maxEventBatchSize | 10 | 単一の呼び出しでバッチに含まれるイベントの最大数。 1 以上を指定してください。 |
| batchCheckpointFrequency | 1 | イベント ハブのチェックポイントを作成する前に処理するバッチの数。 |
| prefetchCount | 300 | Event Hubs から要求され、ローカル キャッシュに保持されるイベントの数。ネットワーク操作の待機を避けるために、読み取りを許可します |
| transportType | amqpTcp | Event Hubs との通信に使用されるプロトコルとトランスポート。 使用可能なオプション: amqpTcp、amqpWebSockets |
| webProxy | Web ソケット経由の Event Hubs との通信に使用されるプロキシ。 プロキシを amqpTcp トランスポートと併用することはできません。 |
|
| customEndpointAddress | Event Hubs への接続を確立するときに使用するアドレス。アプリケーション ゲートウェイまたはホスト環境に必要なその他のパス経由で、ネットワーク要求をルーティングすることができます。 カスタム エンドポイントのアドレスを使用する場合は、イベント ハブの完全修飾名前空間が引き続き必要であり、明示的に、または接続文字列を使用して指定する必要があります。 | |
| initialOffsetOptions/type | fromStart | チェックポイントがストレージに存在しない場合に処理を開始するイベント ストリーム内の位置。 すべてのパーティションに適用されます。 詳細については、OffsetType のドキュメントに関する記事を参照してください。 使用可能なオプション: fromStart、fromEnd、fromEnqueuedTime |
| initialOffsetOptions/enqueuedTimeUtc | ストリームにエンキューされたイベントの処理を開始する時刻を指定します。 initialOffsetOptions/type が fromEnqueuedTime として構成されている場合、この設定は必須です。 DateTime.Parse() でサポートされている形式の時刻がサポートされます ( など)。 明確にするためには、タイムゾーンも指定する必要があります。 タイムゾーンが指定されなかった場合は、関数アプリを実行しているマシンのローカル タイムゾーン (Azure で実行されている場合は UTC) が使用されます。 |
|
| clientRetryOptions/mode | exponential | 再試行の遅延を計算するために使用する方法です。 指数モードでは、バックオフ戦略に基づいて再試行が行われます。この場合、再試行のたびに待機する期間が長くなります。 固定モードでは、一定の間隔で再試行が行われ、遅延の時間はそれぞれ一定です。 使用可能なオプション: exponential、fixed |
| clientRetryOptions/tryTimeout | 00:01:00 | 試行ごとに Event Hubs 操作の完了を待機する最大期間です。 |
| clientRetryOptions/delay | 00:00:00.80 | 再試行の間に適用する遅延またはバックオフ係数です。 |
| clientRetryOptions/maximumDelay | 00:00:01 | 再試行の間の許容される最大の遅延です。 |
| clientRetryOptions/maximumRetries | 3 | 関連する操作が失敗したと判断するまでの再試行の最大回数です。 |
clientRetryOptions は、Functions ホストと Event Hubs の間の操作 (イベントのフェッチやイベントの送信など) を再試行するために使用されます。 個々の関数に再試行ポリシーを適用する方法については、Azure Functions のエラー処理と再試行に関するガイダンスを参照してください。
Azure Functions 2.x 以降の host.json のリファレンスについては、Azure Functions の host.json のリファレンスに関する記事を参照してください。