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)

请分别独立使用各代码片段，并导入 Azure.Messaging.ServiceBus.Administration 以执行 CRUD 操作，导入 System.Transactions 以使用 TransactionScope。确认命名空间/实体以及符合最小权限原则的发送者/接收者/管理员角色。接收操作可能返回 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	消息批次

最佳实践

1. 使用单例 — 客户端、发送器、接收器和处理器均为线程安全
2. 始终释放资源 — 使用 await using 或调用 DisposeAsync()
3. 按顺序释放 — 先关闭发送器/接收器/处理器，再关闭客户端
4. 使用 DefaultAzureCredential — 生产环境中优先于连接字符串
5. 使用处理器执行后台工作 — 自动续订锁
6. 安全地批量发送 — 使用 CreateMessageBatchAsync() 和 TryAddMessage()
7. 处理瞬时错误 — 使用 ServiceBusException.Reason
8. 配置传输方式 — 如果端口 5671/5672 被阻止，请使用 AmqpWebSockets
9. 设置适当的锁持续时间 — 默认值为 30 秒
10. 使用会话保证顺序 — 在单个会话内按 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

参考链接

资源	URL
NuGet 包	https://www.nuget.org/packages/Azure.Messaging.ServiceBus
API 参考	https://learn.microsoft.com/dotnet/api/azure.messaging.servicebus
GitHub 源码	https://github.com/Azure/azure-sdk-for-net/tree/main/sdk/servicebus/Azure.Messaging.ServiceBus
故障排除	https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/servicebus/Azure.Messaging.ServiceBus/TROUBLESHOOTING.md

何时使用

本技能适用于执行概述中描述的工作流或操作。

限制

• 仅在任务明确符合上述范围时使用本技能。
• 请勿将输出视为环境特定验证、测试或专家评审的替代品。
• 如果缺少必需的输入、权限、安全边界或成功标准，请停止操作并请求澄清。