azure-servicebus-dotnet について
azure-servicebus-dotnetは、Azure.Messaging.ServiceBusを扱う.NET開発者向けのスキルです。キューやトピックのメッセージング、処理、セッション、管理、トランザクションの実装方法を示します。
主な機能
- 01メッセージの送受信
- 02バックグラウンドでのメッセージ処理
- 03セッションによる順序処理
- 04デッドレターメッセージの処理
- 05キュー、トピック、サブスクリプションの管理
- 06エンティティ間トランザクションの設定
活用シーン
- 01.NETのキューコンシューマー構築
- 02トピックやサブスクリプションへのメッセージ発行
- 03順序付きセッションメッセージの処理
インストール
コマンドをコピーしてターミナルで実行します。
$
npx skills add https:/インストール後、Claude Code で/azure-servicebus-dotnetを入力して読み込まれたか確認します。
詳細
Azure.Messaging.ServiceBus (.NET)
キュー、トピック、サブスクリプション、セッションを利用してメッセージを確実に配信するためのエンタープライズ向けメッセージング SDK です。
インストール
dotnet add package Azure.Messaging.ServiceBus
dotnet add package Azure.Identity
SDK のバージョン基準例：v7.20.1。スニペットを調整する前に、インストール済みのパッケージと互換性のある API を確認してください。
環境変数
AZURE_SERVICEBUS_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net
# Or connection string (less secure)
AZURE_SERVICEBUS_CONNECTION_STRING=Endpoint=sb://...
認証
Microsoft Entra ID（推奨）
using Azure.Identity;
using Azure.Messaging.ServiceBus;
string fullyQualifiedNamespace = "<namespace>.servicebus.windows.net";
await using ServiceBusClient client = new(fullyQualifiedNamespace, new DefaultAzureCredential());
接続文字列
string connectionString = "<connection_string>";
await using ServiceBusClient client = new(connectionString);
ASP.NET Core の依存関係の挿入
services.AddAzureClients(builder =>
{
builder.AddServiceBusClientWithNamespace("<namespace>.servicebus.windows.net");
builder.UseCredential(new DefaultAzureCredential());
});
クライアントの階層
ServiceBusClient
├── CreateSender(queueOrTopicName) → ServiceBusSender
├── CreateReceiver(queueName) → ServiceBusReceiver
├── CreateReceiver(topicName, subName) → ServiceBusReceiver
├── AcceptNextSessionAsync(queueName) → ServiceBusSessionReceiver
├── CreateProcessor(queueName) → ServiceBusProcessor
└── CreateSessionProcessor(queueName) → ServiceBusSessionProcessor
ServiceBusAdministrationClient (separate client for CRUD)
スニペットはそれぞれ独立して使用し、CRUD には Azure.Messaging.ServiceBus.Administration を、TransactionScope には System.Transactions をインポートしてください。名前空間とエンティティを確認し、最小権限の送信者／受信者／管理者ロールを使用してください。受信結果が null の場合があります。DLQ とトランザクションの例でも同様に null を確認してください。完了するとメッセージは削除されます。まず永続的なビジネス処理を実行し、再配信に備えて冪等にし、結果が不確かな送信や決済を照合してください。すべての例外で放棄すると、有害メッセージのループが発生するおそれがあります。メッセージ本文や、機密性の高いペイロードを含む未加工の例外をログに記録しないでください。セッション対応キューにはセッションレシーバーが必要です。基本的な受信例では、異なるセッション非対応キューを使用しています。プロセッサーは非同期で破棄し、ロックの更新とシャットダウンのポリシーに上限を設けてください。
主なワークフロー
1. メッセージの送信
await using ServiceBusClient client = new(fullyQualifiedNamespace, new DefaultAzureCredential());
ServiceBusSender sender = client.CreateSender("my-queue");
// Single message
ServiceBusMessage message = new("Hello world!");
await sender.SendMessageAsync(message);
// Safe batching (recommended)
using ServiceBusMessageBatch batch = await sender.CreateMessageBatchAsync();
if (!batch.TryAddMessage(new ServiceBusMessage("Message 1")))
throw new InvalidOperationException("Message 1 exceeds the batch capacity");
if (!batch.TryAddMessage(new ServiceBusMessage("Message 2")))
throw new InvalidOperationException("Message 2 exceeds remaining batch capacity; flush and retry in a new batch");
await sender.SendMessagesAsync(batch);
2. メッセージの受信
ServiceBusReceiver receiver = client.CreateReceiver("my-queue");
// Single message
ServiceBusReceivedMessage message = await receiver.ReceiveMessageAsync();
if (message is null) return; // A timed receive can return no message
string body = message.Body.ToString();
Console.WriteLine(body);
// Complete the message (removes from queue)
await receiver.CompleteMessageAsync(message);
// Batch receive
IReadOnlyList<ServiceBusReceivedMessage> messages = await receiver.ReceiveMessagesAsync(maxMessages: 10);
foreach (var msg in messages)
{
Console.WriteLine(msg.Body.ToString());
await receiver.CompleteMessageAsync(msg);
}
3. メッセージの決済
// Complete - removes message from queue
await receiver.CompleteMessageAsync(message);
// Abandon - releases lock, message can be received again
await receiver.AbandonMessageAsync(message);
// Defer - prevents normal receive, use ReceiveDeferredMessageAsync
await receiver.DeferMessageAsync(message);
// Dead Letter - moves to dead letter subqueue
await receiver.DeadLetterMessageAsync(message, "InvalidFormat", "Message body was not valid JSON");
4. プロセッサーを使用したバックグラウンド処理
ServiceBusProcessor processor = client.CreateProcessor("my-queue", new ServiceBusProcessorOptions
{
AutoCompleteMessages = false,
MaxConcurrentCalls = 2
});
processor.ProcessMessageAsync += async (args) =>
{
try
{
string body = args.Message.Body.ToString();
Console.WriteLine($"Received: {body}");
await args.CompleteMessageAsync(args.Message);
}
catch (Exception ex)
{
Console.WriteLine($"Error processing: {ex.Message}");
await args.AbandonMessageAsync(args.Message);
}
};
processor.ProcessErrorAsync += (args) =>
{
Console.WriteLine($"Error source: {args.ErrorSource}");
Console.WriteLine($"Entity: {args.EntityPath}");
Console.WriteLine($"Exception: {args.Exception}");
return Task.CompletedTask;
};
await processor.StartProcessingAsync();
// ... application runs
await processor.StopProcessingAsync();
5. セッション（順序どおりの処理）
// Send session message
ServiceBusMessage message = new("Hello")
{
SessionId = "order-123"
};
await sender.SendMessageAsync(message);
// Receive from next available session
ServiceBusSessionReceiver receiver = await client.AcceptNextSessionAsync("my-queue");
// Or receive from specific session
ServiceBusSessionReceiver receiver = await client.AcceptSessionAsync("my-queue", "order-123");
// Session state management
await receiver.SetSessionStateAsync(new BinaryData("processing"));
BinaryData state = await receiver.GetSessionStateAsync();
// Renew session lock
await receiver.RenewSessionLockAsync();
6. 配信不能キュー
// Receive from dead letter queue
ServiceBusReceiver dlqReceiver = client.CreateReceiver("my-queue", new ServiceBusReceiverOptions
{
SubQueue = SubQueue.DeadLetter
});
ServiceBusReceivedMessage dlqMessage = await dlqReceiver.ReceiveMessageAsync();
// Access dead letter metadata
string reason = dlqMessage.DeadLetterReason;
string description = dlqMessage.DeadLetterErrorDescription;
Console.WriteLine($"Dead letter reason: {reason} - {description}");
7. トピックとサブスクリプション
// Send to topic
ServiceBusSender topicSender = client.CreateSender("my-topic");
await topicSender.SendMessageAsync(new ServiceBusMessage("Broadcast message"));
// Receive from subscription
ServiceBusReceiver subReceiver = client.CreateReceiver("my-topic", "my-subscription");
var message = await subReceiver.ReceiveMessageAsync();
8. 管理（CRUD）
var adminClient = new ServiceBusAdministrationClient(
fullyQualifiedNamespace,
new DefaultAzureCredential());
// Create queue
var options = new CreateQueueOptions("my-queue")
{
MaxDeliveryCount = 10,
LockDuration = TimeSpan.FromSeconds(30),
RequiresSession = true,
DeadLetteringOnMessageExpiration = true
};
QueueProperties queue = await adminClient.CreateQueueAsync(options);
// Update queue
queue.LockDuration = TimeSpan.FromSeconds(60);
await adminClient.UpdateQueueAsync(queue);
// Create topic and subscription
await adminClient.CreateTopicAsync(new CreateTopicOptions("my-topic"));
await adminClient.CreateSubscriptionAsync(new CreateSubscriptionOptions("my-topic", "my-subscription"));
// Delete
await adminClient.DeleteQueueAsync("my-queue");
9. エンティティ間のトランザクション
var options = new ServiceBusClientOptions { EnableCrossEntityTransactions = true };
await using var client = new ServiceBusClient(connectionString, options);
ServiceBusReceiver receiverA = client.CreateReceiver("queueA");
ServiceBusSender senderB = client.CreateSender("queueB");
ServiceBusReceivedMessage receivedMessage = await receiverA.ReceiveMessageAsync();
using (var ts = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
{
await receiverA.CompleteMessageAsync(receivedMessage);
await senderB.SendMessageAsync(new ServiceBusMessage("Forwarded"));
ts.Complete();
}
主な型のリファレンス
| 型 | 用途 |
|---|---|
ServiceBusClient | メインのエントリーポイント。接続を管理します |
ServiceBusSender | キュー／トピックにメッセージを送信します |
ServiceBusReceiver | キュー／サブスクリプションからメッセージを受信します |
ServiceBusSessionReceiver | セッションメッセージを受信します |
ServiceBusProcessor | バックグラウンドでメッセージを処理します |
ServiceBusSessionProcessor | バックグラウンドでセッションメッセージを処理します |
ServiceBusAdministrationClient | キュー／トピック／サブスクリプションの CRUD を行います |
ServiceBusMessage | 送信するメッセージ |
ServiceBusReceivedMessage | メタデータを含む受信済みメッセージ |
ServiceBusMessageBatch | メッセージのバッチ |
ベストプラクティス
- シングルトンを使用する — クライアント、送信者、受信者、プロセッサーはスレッドセーフです
- 必ず破棄する —
await usingを使用するか、DisposeAsync()を呼び出してください - 破棄の順序 — まず送信者／受信者／プロセッサーを閉じ、その後にクライアントを閉じてください
- DefaultAzureCredential を使用する — 本番環境では接続文字列よりも優先してください
- バックグラウンド処理にはプロセッサーを使用する — ロックの更新を自動的に処理します
- 安全にバッチ処理する —
CreateMessageBatchAsync()とTryAddMessage()を使用してください - 一時的なエラーに対処する —
ServiceBusException.Reasonを使用してください - トランスポートを設定する — ポート 5671／5672 がブロックされている場合は
AmqpWebSocketsを使用してください - 適切なロック期間を設定する — 既定値は 30 秒です
- 順序制御にはセッションを使用する — セッション内では FIFO になります
エラー処理
try
{
await sender.SendMessageAsync(message);
}
catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.ServiceBusy)
{
// Retry with backoff
}
catch (ServiceBusException ex)
{
Console.WriteLine($"Service Bus Error: {ex.Reason} - {ex.Message}");
}
関連 SDK
| SDK | 用途 | インストール |
|---|---|---|
Azure.Messaging.ServiceBus | Service Bus（この SDK） | dotnet add package Azure.Messaging.ServiceBus |
Azure.Messaging.EventHubs | イベントストリーミング | dotnet add package Azure.Messaging.EventHubs |
Azure.Messaging.EventGrid | イベントルーティング | dotnet add package Azure.Messaging.EventGrid |
参考リンク
使用するタイミング
このスキルは、概要で説明されているワークフローまたはアクションを実行する場合に適用できます。
制限事項
- このスキルは、タスクが上記のスコープに明確に一致する場合にのみ使用してください。
- 出力を、環境固有の検証、テスト、または専門家によるレビューの代替として扱わないでください。
- 必要な入力、権限、安全境界、または成功基準が欠けている場合は、停止して明確化を求めてください。
コメント
Markdown 対応