メッセージ レプリケーションとクロスリージョン フェデレーションに関する記事で説明されているように、Service Bus エンティティのペア間、および Service Bus と他のメッセージ ソースおよびターゲット間のメッセージ シーケンスのレプリケーションは、通常、Azure Functions に依存しています。
Azure Functions は、メッセージ レプリケーションとフェデレーション タスクを含む、サーバーレス アプリケーションを構成および実行するためのスケーラブルで信頼性の高い実行環境です。
この概要では、このようなアプリケーション用の Azure Functions の組み込み機能、変換タスクに適応して変更できるコード ブロック、および Service Bus と他の Azure Xamarin Messaging サービスと理想的に統合されるように Azure Functions アプリケーションを構成する方法について説明します。 詳細については、この記事で紹介する Azure Functions のドキュメントを参照してください。
レプリケーション タスクとは
ソースからのイベントは、レプリケーション タスクによって受信されて、ターゲットに転送されます。 ほとんどのレプリケーションタスクはイベントを変更せずに転送します。 ソースとターゲットが異なるプロトコルを使用している場合、タスクはそのメタデータ構造をマッピングすることも可能です。
レプリケーションタスクは一般的にステートレスです。 これらは順次実行や並列実行間で状態やその他の副作用を共有しません。 バッチやチェインは、ストリームの既存状態を利用することもできます。
この特性により、レプリケーションタスクは、一般的にステートフルでAzure Stream Analyticsのような分析フレームワークやサービスによってサポートされるアグリゲーションタスクとは異なります。
Azure Functions でのレプリケーション アプリケーションとタスク
Azure Functionsでは、レプリケーションタスクはトリガーを使って送信元からメッセージを受け取ります。 タスクは出力バインディングまたは対応するAzureクライアントライブラリを使ってコピーをターゲットに送信します。
新しい.NETレプリケーションアプリケーションにはAzure Functions .NET孤立ワーカーモデルを活用してください。 .NETのインプロセスモデルのサポートは2026年11月10日に終了します。 既存の進行中のレプリケーションアプリケーションを更新するには、.NETの隔離ワーカー移行ガイドに従ってください。
同じ関数アプリに複数のレプリケーションタスクを展開できます。 Azure Functions Premiumでは、複数の機能アプリが同じApp Serviceプランを共有できます。 この構成により、統合に言語固有のライブラリが必要な場合、異なる言語で書かれたレプリケーションタスクを共配置することも可能です。
バッチ指向のトリガーがあれば、それを好むべきです。 Azure Functionsのバインディング式に頼る代わりに、タスクがソースメタデータを保持できるように、完全なイベントやメッセージ構造を受け取りましょう。
各関数は接続するソースとターゲットの名前を付けます。 接続設定や名前空間の設定には同じ名前をプレフィックスとして使います。
データとメタデータのマッピング
ターゲットがサポートするメッセージ本文とアプリケーションプロパティをマッピングします。 ターゲットは、エンキュー時間やシーケンス番号などのブローカー所有フィールドに新しい値を割り当てます。
repl-enqueue-time および repl-sequence のアプリケーション プロパティでソース値を保持します。 もしどちらかの性質がすでに存在する場合は、新しい値にセミコロン区切り符を付け加えます。 配達数やロック情報はコピーしないでください。
アイソレーテッドワーカー Service Bus出力バインディングは単純な出力タイプをサポートしますが、ServiceBusMessageはサポートしません。
ServiceBusClient は、レプリケーション タスクで Service Bus メッセージのメタデータを保持する必要がある場合に使用します。 クライアントを使うことで、コードが送信失敗を処理し、ログを記録できます。 以下の例では、この動作を明示的にするために、両方のターゲットに対してAzureクライアントライブラリを使用しています。
再試行ポリシー
再試行はソーストリガーとターゲットクライアントに基づいて設定します。 Event Hubsトリガーについては、Azure Functionsの関数レベルの再試行ポリシーを適用できます。 Event Hubsは実行の再試行ポリシーが完了するまでチェックポイントを書きません。
Service Busトリガーはエンティティおよびhost.json上で設定されたリトライおよびデッドレターの動作を利用します。 Service Busトリガーに関数レベルのretry attributeを適用しないでください。 失敗したバッチは全額再配達され、ターゲットで重複が生まれる可能性があります。 メッセージが送信元のエンティティのMaxDeliveryCountを超えると、Service Busはそれをデッドレターキューに移動します。 必要な回復ウィンドウに MaxDeliveryCount を設定し、デッドレターキューを監視します。
Azureクライアントライブラリも送信操作に設定された再試行ポリシーを適用します。 それでも送信が失敗する場合は、例外を関数から脱出させ、ソーストリガーがバッチをやり直せるようにします。 Event Hubsの再試行ポリシーは無制限の再試行数を持ち、対象の送信が成功するまで影響を受けたパーティションのチェックポイント進行を一時停止します。
レプリケーションアプリケーションホストをセットアップする
レプリケーションアプリケーションとは、1つ以上のレプリケーションタスクをホストするAzure Functionsアプリケーションです。 Functions 4.xランタイムのサポート版と.NETの孤立ワーカーモデルを使用してください。 規模やネットワークの要件に合ったプランでアプリを作成しましょう。
機能アプリには システム割り当てまたはユーザー割り当てのマネージドID を使用してください。 アイデンティティに、各ソースから受信する権限と各ターゲットに送信する権限を付与します。 トリガー用のアイデンティティベースの接続を設定し、Azureクライアントの登録に対して完全に限定されたターゲット名前空間を提供します。 例としては、 telemetrySourceConnection__fullyQualifiedNamespace、 telemetryTarget__fullyQualifiedNamespace、 jobsTransferSourceConnection__fullyQualifiedNamespace、 jobsTransferTarget__fullyQualifiedNamespaceの集合です。
以下の Program.cs 例では、 DefaultAzureCredentialを用いてターゲットサービス接続を登録します。 Azureでは、DefaultAzureCredential関数アプリのマネージドIDを利用しています。
using Azure.Identity;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Azure;
using Microsoft.Extensions.Hosting;
var builder = FunctionsApplication.CreateBuilder(args);
builder.Services.AddAzureClients(clientBuilder =>
{
clientBuilder.AddEventHubProducerClientWithNamespace(
builder.Configuration["telemetryTarget:fullyQualifiedNamespace"],
"telemetry-copy");
clientBuilder.AddServiceBusClientWithNamespace(
builder.Configuration["jobsTransferTarget:fullyQualifiedNamespace"]);
clientBuilder.UseCredential(new DefaultAzureCredential());
});
builder.Build().Run();
プロジェクトには分離されたワーカーパッケージ、イベントハブ、Service Busバインディング拡張、Azure.Identity、Azure.Messaging.EventHubs、Azure.Messaging.ServiceBus、Microsoft.Extensions.Azureが必要です。 現在のパッケージおよびターゲットフレームワークの要件については、.NET isolated worker guideおよびmigration guideをご覧ください。
仮想ネットワークを通じてEvent HubsやService Busの名前空間にアクセスするレプリケーションアプリケーションは、仮想ネットワーク統合をサポートするホスティングプランを使用しなければなりません。 詳細については、「Azure Functions ネットワーク オプションを参照してください。
例
以下の.NET孤立したワーカーの例は、バッチでメッセージを受け取り、顧客が管理するボディとメタデータをコピーし、Program.csに登録されたターゲットクライアントを使用します。
イベントハブ間でイベントデータをコピーするには、Event Hubsトリガーと EventHubProducerClientを使用します。
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;
using Microsoft.Azure.Functions.Worker;
public sealed class EventHubReplicationFunction
{
private readonly EventHubProducerClient outputClient;
public EventHubReplicationFunction(EventHubProducerClient outputClient)
{
this.outputClient = outputClient;
}
[Function("telemetry")]
[ExponentialBackoffRetry(-1, "00:00:05", "00:15:00")]
public async Task Telemetry(
[EventHubTrigger(
"telemetry",
ConsumerGroup = "%telemetrySourceConsumerGroup%",
Connection = "telemetrySourceConnection")]
EventData[] input,
CancellationToken cancellationToken)
{
foreach (IGrouping<string, EventData> group in input.GroupBy(item => item.PartitionKey))
{
var options = new CreateBatchOptions { PartitionKey = group.Key };
EventDataBatch batch = await outputClient.CreateBatchAsync(options, cancellationToken);
try
{
foreach (EventData item in group)
{
EventData copy = CopyEvent(item);
if (!batch.TryAdd(copy))
{
if (batch.Count == 0)
{
throw new InvalidOperationException("An event exceeds the maximum Event Hubs batch size.");
}
await outputClient.SendAsync(batch, cancellationToken);
batch.Dispose();
batch = await outputClient.CreateBatchAsync(options, cancellationToken);
if (!batch.TryAdd(copy))
{
throw new InvalidOperationException("An event exceeds the maximum Event Hubs batch size.");
}
}
}
if (batch.Count > 0)
{
await outputClient.SendAsync(batch, cancellationToken);
}
}
finally
{
batch.Dispose();
}
}
}
private static EventData CopyEvent(EventData input)
{
var output = new EventData(input.EventBody)
{
ContentType = input.ContentType,
CorrelationId = input.CorrelationId,
MessageId = input.MessageId,
};
foreach (KeyValuePair<string, object> property in input.Properties)
{
output.Properties[property.Key] = property.Value;
}
AppendReplicationProperty(
output.Properties,
"repl-enqueue-time",
FormatSystemProperty(input, "x-opt-enqueued-time"));
AppendReplicationProperty(
output.Properties,
"repl-sequence",
FormatSystemProperty(input, "x-opt-sequence-number"));
return output;
}
private static string FormatSystemProperty(EventData input, string name)
{
if (!input.SystemProperties.TryGetValue(name, out object? value))
{
throw new InvalidOperationException($"Event Hubs didn't provide {name}.");
}
return value switch
{
DateTime dateTime => dateTime.ToUniversalTime().ToString("O"),
DateTimeOffset dateTimeOffset => dateTimeOffset.ToUniversalTime().ToString("O"),
IFormattable formattable => formattable.ToString(null, CultureInfo.InvariantCulture),
_ => value.ToString()!,
};
}
private static void AppendReplicationProperty(
IDictionary<string, object> properties,
string name,
string value)
{
properties[name] = properties.TryGetValue(name, out object? existing)
? $"{existing};{value}"
: value;
}
}
Service Busエンティティ間でメッセージをコピーするには、Service BusトリガーとServiceBusClientを使用します。
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker;
public sealed class ServiceBusReplicationFunction
{
private readonly ServiceBusSender sender;
public ServiceBusReplicationFunction(ServiceBusClient client)
{
sender = client.CreateSender("jobs");
}
[Function("jobs-transfer")]
public async Task JobsTransfer(
[ServiceBusTrigger(
"jobs-transfer",
Connection = "jobsTransferSourceConnection",
IsBatched = true,
IsSessionsEnabled = true)]
ServiceBusReceivedMessage[] input,
CancellationToken cancellationToken)
{
foreach (IGrouping<(string SessionId, string PartitionKey), ServiceBusReceivedMessage> group
in input.GroupBy(item => (item.SessionId, item.PartitionKey)))
{
ServiceBusMessageBatch batch = await sender.CreateMessageBatchAsync(cancellationToken);
try
{
foreach (ServiceBusReceivedMessage item in group)
{
ServiceBusMessage copy = CopyMessage(item);
if (!batch.TryAddMessage(copy))
{
if (batch.Count == 0)
{
throw new InvalidOperationException("A message exceeds the maximum Service Bus batch size.");
}
await sender.SendMessagesAsync(batch, cancellationToken);
batch.Dispose();
batch = await sender.CreateMessageBatchAsync(cancellationToken);
if (!batch.TryAddMessage(copy))
{
throw new InvalidOperationException("A message exceeds the maximum Service Bus batch size.");
}
}
}
if (batch.Count > 0)
{
await sender.SendMessagesAsync(batch, cancellationToken);
}
}
finally
{
batch.Dispose();
}
}
}
private static ServiceBusMessage CopyMessage(ServiceBusReceivedMessage input)
{
var output = new ServiceBusMessage(input.Body)
{
ContentType = input.ContentType,
CorrelationId = input.CorrelationId,
MessageId = input.MessageId,
PartitionKey = input.PartitionKey,
ReplyTo = input.ReplyTo,
ReplyToSessionId = input.ReplyToSessionId,
SessionId = input.SessionId,
Subject = input.Subject,
To = input.To,
};
foreach (KeyValuePair<string, object> property in input.ApplicationProperties)
{
output.ApplicationProperties[property.Key] = property.Value;
}
AppendReplicationProperty(
output.ApplicationProperties,
"repl-enqueue-time",
input.EnqueuedTime.ToString("O"));
AppendReplicationProperty(
output.ApplicationProperties,
"repl-sequence",
input.SequenceNumber.ToString(CultureInfo.InvariantCulture));
return output;
}
private static void AppendReplicationProperty(
IDictionary<string, object> properties,
string name,
string value)
{
properties[name] = properties.TryGetValue(name, out object? existing)
? $"{existing};{value}"
: value;
}
}
Service Busの例では、セッションIDとパーティションキーでメッセージをグループ化し、各グループをソース順に送信します。 このアプローチは相対的なセッション順序を保持します。 どちらの例も、ターゲットサービスの最大バッチサイズ内に収まるサイズ制限バッチを作成します。 Service Busの例では、元の生存時間をコピーしません。なぜなら、コピーするとターゲットのライフタイムがリセットされるからです。 有効期限認識の複製では、残りの寿命を計算するか、対象エンティティの有効期限ポリシーを適用してください。
本番環境のレプリケーションパスでは、重複配信、メッセージの期限切れ、トランザクション、スケジュールされたメッセージ、死線メッセージ、ターゲットプロトコルが表現できないメタデータの処理方法も決定します。
監視
Azure Functions monitoringを使ってレプリケーションアプリケーションを監視してください。
Application Insights Application Map は、レプリケーションタスクとそのソースおよびターゲット間の依存関係を可視化します。 Live Metricsは 、関数アプリが動作している間に低遅延で診断情報を提供します。