azure-servicebus-dotnet

作者: microsoft

Azure Service Bus SDK for .NET。企业级消息传递,支持队列、主题、订阅和会话。用于可靠消息投递、发布/订阅模式、死信处理和后台处理。触发词:“Service Bus”、“ServiceBusClient”、“ServiceBusSender”、“ServiceBusReceiver”、“ServiceBusProcessor”、“消息队列”、“pub/sub .NET”、“死信队列”。

npx skills add https://github.com/microsoft/skills --skill azure-servicebus-dotnet

Azure.Messaging.ServiceBus (.NET)

Enterprise messaging SDK for reliable message delivery with queues, topics, subscriptions, and sessions.

Installation

dotnet add package Azure.Messaging.ServiceBus
dotnet add package Azure.Identity

Current Version: v7.20.1 (stable)

Environment Variables

AZURE_SERVICEBUS_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net  # Required: Service Bus fully qualified namespace
AZURE_TOKEN_CREDENTIALS=prod  # Required only if DefaultAzureCredential is used in production
AZURE_SERVICEBUS_CONNECTION_STRING=Endpoint=sb://...  # Alternative to Entra ID auth

Authentication

Microsoft Entra Token Credential

using Azure.Identity;
using Azure.Messaging.ServiceBus;

string fullyQualifiedNamespace = "<namespace>.servicebus.windows.net";
// Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
var credential = new DefaultAzureCredential(
    DefaultAzureCredential.DefaultEnvironmentVariableName
);
// Or use a specific credential directly in production:
// See https://learn.microsoft.com/dotnet/api/overview/azure/identity-readme?view=azure-dotnet#credential-classes
// var credential = new ManagedIdentityCredential();
await using ServiceBusClient client = new(fullyQualifiedNamespace, credential);

Connection String

string connectionString = "<connection_string>";
await using ServiceBusClient client = new(connectionString);

ASP.NET Core Dependency Injection

services.AddAzureClients(builder =>
{
    builder.AddServiceBusClientWithNamespace("<namespace>.servicebus.windows.net");
    builder.UseCredential(new DefaultAzureCredential());
});

Client Hierarchy

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)

Core Workflows

1. Send Messages

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")))
{
    // Message added successfully
}
if (batch.TryAddMessage(new ServiceBusMessage("Message 2")))
{
    // Message added successfully
}
await sender.SendMessagesAsync(batch);

2. Receive Messages

ServiceBusReceiver receiver = client.CreateReceiver("my-queue");

// Single message
ServiceBusReceivedMessage message = await receiver.ReceiveMessageAsync();
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. Message Settlement

// 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. Background Processing with Processor

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. Sessions (Ordered Processing)

// 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. Dead Letter Queue

// 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. Topics and Subscriptions

// 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. Administration (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. Cross-Entity Transactions

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();
}

Key Types Reference

TypePurpose
ServiceBusClientMain entry point, manages connection
ServiceBusSenderSends messages to queues/topics
ServiceBusReceiverReceives messages from queues/subscriptions
ServiceBusSessionReceiverReceives session messages
ServiceBusProcessorBackground message processing
ServiceBusSessionProcessorBackground session processing
ServiceBusAdministrationClientCRUD for queues/topics/subscriptions
ServiceBusMessageMessage to send
ServiceBusReceivedMessageReceived message with metadata
ServiceBusMessageBatchBatch of messages

Best Practices

  1. Use singletons — Clients, senders, receivers, and processors are thread-safe
  2. Always dispose — Use await using or call DisposeAsync()
  3. Dispose order — Close senders/receivers/processors first, then client
  4. Use DefaultAzureCredential — Prefer over connection strings for production
  5. Use processors for background work — Handles lock renewal automatically
  6. Use safe batching — CreateMessageBatchAsync() and TryAddMessage()
  7. Handle transient errors — Use ServiceBusException.Reason
  8. Configure transport — Use AmqpWebSockets if ports 5671/5672 are blocked
  9. Set appropriate lock duration — Default is 30 seconds
  10. Use sessions for ordering — FIFO within a session

Error Handling

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}");
}

Related SDKs

SDKPurposeInstall
Azure.Messaging.ServiceBusService Bus (this SDK)dotnet add package Azure.Messaging.ServiceBus
Azure.Messaging.EventHubsEvent streamingdotnet add package Azure.Messaging.EventHubs
Azure.Messaging.EventGridEvent routingdotnet add package Azure.Messaging.EventGrid

Reference Links

ResourceURL
NuGet Packagehttps://www.nuget.org/packages/Azure.Messaging.ServiceBus
API Referencehttps://learn.microsoft.com/dotnet/api/azure.messaging.servicebus
GitHub Sourcehttps://github.com/Azure/azure-sdk-for-net/tree/main/sdk/servicebus/Azure.Messaging.ServiceBus
Troubleshootinghttps://github.com/Azure/azure-sdk-for-net/blob/main/sdk/servicebus/Azure.Messaging.ServiceBus/TROUBLESHOOTING.md

来自 microsoft 的更多技能

oss-growth
microsoft
OSS增长黑客角色
agent-framework-azure-ai-py
microsoft
使用Microsoft Agent Framework Python SDK(agent-framework-azure-ai)构建Azure AI Foundry代理。在创建使用AzureAIAgentsProvider的持久化代理、使用托管工具(代码解释器、文件搜索、网络搜索)、集成MCP服务器、管理对话线程或实现流式响应时使用。涵盖函数工具、结构化输出和多工具代理。
development
airunway-aks-setup
microsoft
在AKS上设置AI Runway——从裸集群到运行模型。涵盖集群验证、控制器安装、GPU评估、提供商设置和首次部署。适用场景:“设置AI Runway”、“接入AKS集群”、“安装AI Runway”、“airunway设置”、“将模型部署到AKS”、“在AKS上进行GPU推理”、“在AKS上配置KAITO”、“在AKS上运行LLM”、“在AKS上使用vLLM”、“在AKS上设置模型服务”、“AI Runway控制器”。
devops
appinsights-instrumentation
microsoft
使用Azure Application Insights对Web应用进行插桩的指南。提供遥测模式、SDK设置和配置参考。适用场景:如何对应用进行插桩、App Insights SDK、遥测模式、什么是App Insights、Application Insights指南、插桩示例、APM最佳实践。
devops
applicationinsights-web-ts
microsoft
使用Application Insights JavaScript SDK(@microsoft/applicationinsights-web)为浏览器/Web应用添加检测。用于真实用户监控(RUM)——页面视图、点击、AJAX/fetch依赖项、异常、自定义事件,以及与后端OpenTelemetry追踪关联的浏览器端GenAI代理追踪。涵盖SDK加载器脚本和npm设置、框架扩展(React、React Native、Angular)、点击分析、遥测初始化器,以及从浏览器发出的代理/工具/模型跨度所遵循的OTel GenAI语义约定。
devops
azure-ai-anomalydetector-java
microsoft
使用适用于 Java 的 Azure AI 异常检测器 SDK 构建异常检测应用程序。在实现单变量/多变量异常检测、时间序列分析或 AI 驱动的监控时使用。
development
azure-ai-language-conversations-py
microsoft
使用azure-ai-language-conversations Python SDK实现对话语言理解(CLU)。当使用ConversationAnalysisClient分析对话意图和实体、构建NLP功能或将语言理解集成到应用程序中时使用。
development
azure-ai-ml-py
microsoft
Azure Machine Learning SDK v2 for Python。用于机器学习工作区、作业、模型、数据集、计算资源和管道。 触发词:“azure-ai-ml”、“MLClient”、“工作区”、“模型注册表”、“训练作业”、“数据集”。
development