Azure.Messaging.ServiceBus 7.18.1

Azure服务总线.NET客户端库

Azure服务总线允许您构建可以利用异步消息模式的应用程序，它通过高度可靠的服务在生产者和消费者之间转发消息。Azure服务总线为客户端和服务器之间提供灵活的代理消息，以及结构化先进先出（FIFO）消息和具有复杂路由的发布/订阅功能。如果您想了解更多关于Azure服务总线的信息，您可能希望查看：什么是Azure服务总线？

使用Azure服务总线客户端库来完成以下操作：

• 传输业务数据：通过消息利用持久交换信息，例如销售或采购订单、日记或库存变动。

• 解耦应用程序：提高应用程序和服务的可靠性和可扩展性，减轻发送者和接收者同时在线的需要。

• 控制消息的处理方式：支持使用队列的传统竞争型消费者，或允许每个消费者使用主题和订阅拥有自己的消息实例。

• 实现复杂的工作流：消息会话支持需要消息排序或消息延迟的场景。

源代码 | 包 (NuGet) | API 参考文档 | 产品文档 | 迁移指南 (Microsoft.Azure.ServiceBus) | 迁移指南 (WindowsAzure.ServiceBus) | 故障排除指南

入门指南

先决条件

• Microsoft Azure 订阅：使用 Azure 服务，包括 Azure Service Bus，您需要一个订阅。如果您没有现有的 Azure 帐户，您可以在创建帐户时注册免费试用或使用您的 MSDN 订阅者权益。

• 服务总署名称空间：要与服务总署进行交互，您还需要一个可用的名称空间。如果您不熟悉创建 Azure 资源，您可能希望遵循使用 Azure 门户创建 Service Bus 名称空间的分步指南。在那里，您也可以找到使用 Azure CLI、Azure PowerShell 或 Azure 资源管理器 (ARM) 模板创建 Service Bus 实体的详细说明。

• C# 8.0： Azure Service Bus 客户端库利用了 C# 8.0 中引入的新功能。为了利用 C# 8.0 语法，建议您使用 3.0 或更高版本的 .NET Core SDK 并设置 语言版本 为 latest 来进行编译。

  希望充分利用 C# 8.0 语法的 Visual Studio 用户需要使用 Visual Studio 2019 或更高版本。您可以在此处下载 Visual Studio 2019，包括免费的社区版。Visual Studio 2017 用户可以通过使用 Microsoft.Net.Compilers NuGet 包 并设置语言版本来利用 C# 8 语法，尽管编辑体验可能并不理想。

  您仍然可以使用库与之前的 C# 语言版本，但需要手动管理异步可枚举和异步可丢弃成员，而不是从新语法中受益。您仍然可以将目标框架版本设置为您 .NET Core SDK 支持的任何版本，包括更早版本的 .NET Core 或 .NET 框架。有关更多信息，请参阅：如何指定目标框架。

  重要说明：为了无需修改就构建或运行示例和样例，强制使用 C# 8.0。如果您决定对其他语言版本进行调整，仍可以运行样例。

要快速在 Azure 中创建需要的 Service Bus 资源并获得它们的连接字符串，您可以单击

安装包

使用 NuGet 安装 Azure Service Bus 客户端库 for .NET

dotnet add package Azure.Messaging.ServiceBus

认证客户端

为了使服务总线客户端库能够与队列或主题交互，它需要了解如何与之建立连接和授权。最容易的方法是使用连接字符串，它是在创建服务总线命名空间时自动创建的。如果您对Azure中的共享访问策略不熟悉，您可能希望遵循获取服务总线连接字符串的步骤指南。

一旦您有了连接字符串，就可以使用它对客户端进行身份验证。

// Create a ServiceBusClient that will authenticate using a connection string
string connectionString = "<connection_string>";
await using var client = new ServiceBusClient(connectionString);

要了解如何使用Azure.Identity进行身份验证，请参阅此示例。

要了解如何为ASP.NET Core应用程序进行身份验证，请参阅此示例。

要了解如何使用自定义端点初始化连接，请参阅此示例。

关键概念

初始化ServiceBusClient后，您可以与服务总线命名空间内的主要资源类型进行交互，其中可以存在多个此类命名空间，实际的消息传输在这里进行，命名空间通常作为应用程序的容器。

• 队列：允许发送和接收消息。通常用于点对点通信。

• 主题：与队列不同，主题更适合发布/订阅场景。主题可以被发送，但需要一个订阅来消费，可以有多个并行订阅来消费。

• 订阅：从主题中消费的机制。每个订阅都是独立的，并接收发送到主题的每个消息的副本。可以使用规则和过滤器来调整特定订阅接收到的哪些消息。

有关这些资源的更多信息，请参阅什么是Azure服务总线？。

要与这些资源进行交互，应熟悉以下SDK概念

• 一个服务总线客户端是开发人员与服务总线客户端库交互的主要接口。它作为所有与库交互的网关。

• 一个服务总线发送者局限于特定的队列或主题，并使用服务总线客户端创建。发送者允许您向队列或主题发送消息。它还允许为在指定日期可用的消息进行排程。

• 一个服务总线接收者局限于特定的队列或订阅，并使用服务总线客户端创建。接收者允许您从队列或订阅中接收消息。它还允许在接收后解决消息。有四种解决消息的方式

  • 完成 - 将消息从队列或主题中删除。
  • 放弃 - 释放接收者对消息的锁定，允许其他接收者接收该消息。
  • 延迟 - 延迟消息以正常方式接收。为了接收延迟消息，需要保留消息的序列号。
  • 死信队列 - 将消息移至死信队列。这将防止消息再次被接收。为了从死信队列接收消息，需要一个与死信队列相关的接收器。

• 服务总线会话接收器被限制在特定的会话启用队列或订阅中，并使用服务总线客户端创建。会话接收器几乎与标准接收器相同，不同之处在于暴露了仅适用于会话启用实体的会话管理操作。这些操作包括获取和设置会话状态，以及续订会话锁。

• 服务总线处理器被限制在特定的队列或订阅中，并使用服务总线客户端创建。《ServiceBusProcessor》可以被看作是一系列接收器的抽象。它使用回调模型允许指定在接收到消息和发生异常时执行的操作。它提供已处理消息的自动完成、自动消息锁续订以及并发执行用户指定的事件处理器。由于其功能集，应该将其作为编写从服务总线实体接收应用的首选工具。当处理器无法提供使用 ServiceBusReceiver 直接收到的精细控制时，推荐使用 ServiceBusReceiver 来处理更复杂的场景。

• 服务总线会话处理器被限制在特定的会话启用队列或订阅中，并使用服务总线客户端创建。会话处理器几乎与标准处理器相同，不同之处在于暴露了仅适用于会话启用实体的会话管理操作。

有关更多概念和深入讨论，请参阅： 服务总线高级功能。

客户端生命周期

《ServiceBusClient》、发送者、接收器和处理器都可以安全地进行缓存，并在整个应用生命周期中作为单例使用，这在使用消息时是最佳实践。它们负责有效管理网络、CPU 和内存使用，努力在空闲期间保持低使用率。

这些类型是可丢弃的，调用 DisposeAsync 或 CloseAsync 是必不可少的，以确保正确清理网络资源和其他非托管对象。重要的是要注意，当《ServiceBusClient》实例被垃圾回收时，基础 AMQP 连接关闭，因此使用该实例创建的任何发送器、接收器和处理器都无法再使用，无论发送器、接收器和处理器是否被显式关闭。最佳实践是在关闭发送器、接收器和处理器确保清理 AMQP 链接后，再关闭《ServiceBusClient》以确保 AMQP 连接关闭。

线程安全

我们保证所有客户端实例方法都是线程安全的并且彼此独立（指南）。这确保了即使在多线程的情况下，建议重用客户端实例始终是安全的。

其他概念

客户端选项 | 诊断 | Mocking

示例

• Azure服务总线.NET客户端库
  • 入门指南
    • 先决条件
    • 安装包
    • 认证客户端
  • 关键概念
    • 客户端生命周期
    • 线程安全
    • 其他概念
  • 示例
    • 发送和接收消息
    • 发送一批消息
    • 接收一批消息
    • 完成消息
    • 放弃消息
    • 延迟消息
    • 死信消息
    • 使用处理器
    • 使用 Azure.Identity 进行身份验证
    • 使用 ASP.NET Core 依赖注入进行注册
    • 使用会话
  • 故障排除
  • 下一步
  • 贡献

发送和接收消息

消息发送使用 ServiceBusSender 进行。接收使用 ServiceBusReceiver 进行。

string connectionString = "<connection_string>";
string queueName = "<queue_name>";
// since ServiceBusClient implements IAsyncDisposable we create it with "await using"
await using var client = new ServiceBusClient(connectionString);

// create the sender
ServiceBusSender sender = client.CreateSender(queueName);

// create a message that we can send. UTF-8 encoding is used when providing a string.
ServiceBusMessage message = new ServiceBusMessage("Hello world!");

// send the message
await sender.SendMessageAsync(message);

// create a receiver that we can use to receive the message
ServiceBusReceiver receiver = client.CreateReceiver(queueName);

// the received message is a different type as it contains some service set properties
ServiceBusReceivedMessage receivedMessage = await receiver.ReceiveMessageAsync();

// get the message body as a string
string body = receivedMessage.Body.ToString();
Console.WriteLine(body);

发送一批消息

有几种方式可以同时发送多条消息。第一种方式使用安全批量处理。在安全批量处理中，你可以创建一个 ServiceBusMessageBatch 对象，这将允许你使用 TryAdd 方法一次向批次添加一条消息。如果消息不能放入批次，TryAdd 将返回 false。

// add the messages that we plan to send to a local queue
Queue<ServiceBusMessage> messages = new Queue<ServiceBusMessage>();
messages.Enqueue(new ServiceBusMessage("First message"));
messages.Enqueue(new ServiceBusMessage("Second message"));
messages.Enqueue(new ServiceBusMessage("Third message"));

// create a message batch that we can send
// total number of messages to be sent to the Service Bus queue
int messageCount = messages.Count;

// while all messages are not sent to the Service Bus queue
while (messages.Count > 0)
{
    // start a new batch
    using ServiceBusMessageBatch messageBatch = await sender.CreateMessageBatchAsync();

    // add the first message to the batch
    if (messageBatch.TryAddMessage(messages.Peek()))
    {
        // dequeue the message from the .NET queue once the message is added to the batch
        messages.Dequeue();
    }
    else
    {
        // if the first message can't fit, then it is too large for the batch
        throw new Exception($"Message {messageCount - messages.Count} is too large and cannot be sent.");
    }

    // add as many messages as possible to the current batch
    while (messages.Count > 0 && messageBatch.TryAddMessage(messages.Peek()))
    {
        // dequeue the message from the .NET queue as it has been added to the batch
        messages.Dequeue();
    }

    // now, send the batch
    await sender.SendMessagesAsync(messageBatch);

    // if there are any remaining messages in the .NET queue, the while loop repeats
}

第二种方式使用接受 ServiceBusMessage 的 IEnumerable 的 SendMessagesAsync 重载。使用此方法，我们将尝试将所有提供的消息放入一个我们将发送给服务的单个消息批次中。如果消息太大而无法放入单个批次，操作将引发异常。

IList<ServiceBusMessage> messages = new List<ServiceBusMessage>();
messages.Add(new ServiceBusMessage("First"));
messages.Add(new ServiceBusMessage("Second"));
// send the messages
await sender.SendMessagesAsync(messages);

接收一批消息

// create a receiver that we can use to receive the messages
ServiceBusReceiver receiver = client.CreateReceiver(queueName);

// the received message is a different type as it contains some service set properties
// a batch of messages (maximum of 2 in this case) are received
IReadOnlyList<ServiceBusReceivedMessage> receivedMessages = await receiver.ReceiveMessagesAsync(maxMessages: 2);

// go through each of the messages received
foreach (ServiceBusReceivedMessage receivedMessage in receivedMessages)
{
    // get the message body as a string
    string body = receivedMessage.Body.ToString();
}

完成消息

为了从队列或订阅中移除消息，我们可以调用 CompleteMessageAsync 方法。

string connectionString = "<connection_string>";
string queueName = "<queue_name>";
// since ServiceBusClient implements IAsyncDisposable we create it with "await using"
await using var client = new ServiceBusClient(connectionString);

// create the sender
ServiceBusSender sender = client.CreateSender(queueName);

// create a message that we can send
ServiceBusMessage message = new ServiceBusMessage("Hello world!");

// send the message
await sender.SendMessageAsync(message);

// create a receiver that we can use to receive and settle the message
ServiceBusReceiver receiver = client.CreateReceiver(queueName);

// the received message is a different type as it contains some service set properties
ServiceBusReceivedMessage receivedMessage = await receiver.ReceiveMessageAsync();

// complete the message, thereby deleting it from the service
await receiver.CompleteMessageAsync(receivedMessage);

放弃消息

放弃消息将释放我们的接收器锁，这允许该消息被该接收器或其他接收器接收。

ServiceBusReceivedMessage receivedMessage = await receiver.ReceiveMessageAsync();

// abandon the message, thereby releasing the lock and allowing it to be received again by this or other receivers
await receiver.AbandonMessageAsync(receivedMessage);

延迟消息

延迟消息将防止它通过 ReceiveMessageAsync 或 ReceiveMessagesAsync 方法再次接收。相反，有分离的方法，ReceiveDeferredMessageAsync 和 ReceiveDeferredMessagesAsync 用于接收延迟消息。

ServiceBusReceivedMessage receivedMessage = await receiver.ReceiveMessageAsync();

// defer the message, thereby preventing the message from being received again without using
// the received deferred message API.
await receiver.DeferMessageAsync(receivedMessage);

// receive the deferred message by specifying the service set sequence number of the original
// received message
ServiceBusReceivedMessage deferredMessage = await receiver.ReceiveDeferredMessageAsync(receivedMessage.SequenceNumber);

死信消息

死信消息类似于延迟，主要区别在于服务将在消息接收一定次数后将消息自动死信处理。应用程序可以根据其需求选择手动死信处理消息。消息被死信处理后，实际上是被移动到原始队列的子队列中。请注意，无论主队列是否启用会话，都使用 ServiceBusReceiver 从死信子队列接收消息。

ServiceBusReceivedMessage receivedMessage = await receiver.ReceiveMessageAsync();

// Dead-letter the message, thereby preventing the message from being received again without receiving from the dead letter queue.
// We can optionally pass a dead letter reason and dead letter description to further describe the reason for dead-lettering the message.
await receiver.DeadLetterMessageAsync(receivedMessage, "sample reason", "sample description");

// receive the dead lettered message with receiver scoped to the dead letter queue.
ServiceBusReceiver dlqReceiver = client.CreateReceiver(queueName, new ServiceBusReceiverOptions
{
    SubQueue = SubQueue.DeadLetter
});
ServiceBusReceivedMessage dlqMessage = await dlqReceiver.ReceiveMessageAsync();

// The reason and the description that we specified when dead-lettering the message will be available in the received dead letter message.
string reason = dlqMessage.DeadLetterReason;
string description = dlqMessage.DeadLetterErrorDescription;

有关更多信息，请参阅 ServiceBus 死信队列概述。

使用处理器

ServiceBusProcessor 可以被视为围绕一组接收器的一组抽象。它使用回调模型，在收到消息和发生异常时允许指定代码。它提供自动完成处理消息、自动消息锁定更新和执行用户指定的事件处理器的并发执行。由于其功能集，它应该是编写从 Service Bus 实体接收应用程序的首选工具。对于不提供使用 ServiceBusReceiver 直接时可以期望的精细控制的情况，建议使用 ServiceBusReceiver。

string connectionString = "<connection_string>";
string queueName = "<queue_name>";
// since ServiceBusClient implements IAsyncDisposable we create it with "await using"
await using var client = new ServiceBusClient(connectionString);

// create the sender
ServiceBusSender sender = client.CreateSender(queueName);

// create a set of messages that we can send
ServiceBusMessage[] messages = new ServiceBusMessage[]
{
    new ServiceBusMessage("First"),
    new ServiceBusMessage("Second")
};

// send the message batch
await sender.SendMessagesAsync(messages);

// create the options to use for configuring the processor
var options = new ServiceBusProcessorOptions
{
    // By default or when AutoCompleteMessages is set to true, the processor will complete the message after executing the message handler
    // Set AutoCompleteMessages to false to [settle messages](https://docs.microsoft.com/en-us/azure/service-bus-messaging/message-transfers-locks-settlement#peeklock) on your own.
    // In both cases, if the message handler throws an exception without settling the message, the processor will abandon the message.
    AutoCompleteMessages = false,

    // I can also allow for multi-threading
    MaxConcurrentCalls = 2
};

// create a processor that we can use to process the messages
await using ServiceBusProcessor processor = client.CreateProcessor(queueName, options);

// configure the message and error handler to use
processor.ProcessMessageAsync += MessageHandler;
processor.ProcessErrorAsync += ErrorHandler;

async Task MessageHandler(ProcessMessageEventArgs args)
{
    string body = args.Message.Body.ToString();
    Console.WriteLine(body);

    // we can evaluate application logic and use that to determine how to settle the message.
    await args.CompleteMessageAsync(args.Message);
}

Task ErrorHandler(ProcessErrorEventArgs args)
{
    // the error source tells me at what point in the processing an error occurred
    Console.WriteLine(args.ErrorSource);
    // the fully qualified namespace is available
    Console.WriteLine(args.FullyQualifiedNamespace);
    // as well as the entity path
    Console.WriteLine(args.EntityPath);
    Console.WriteLine(args.Exception.ToString());
    return Task.CompletedTask;
}

// start processing
await processor.StartProcessingAsync();

// since the processing happens in the background, we add a Console.ReadKey to allow the processing to continue until a key is pressed.
Console.ReadKey();

使用 Azure.Identity 进行身份验证

Azure Identity 库 为身份验证提供了简单的 Azure Active Directory 支持。

// Create a ServiceBusClient that will authenticate through Active Directory
string fullyQualifiedNamespace = "yournamespace.servicebus.windows.net";
await using var client = new ServiceBusClient(fullyQualifiedNamespace, new DefaultAzureCredential());

使用会话

会话 为分组相关消息提供了一个机制。为了使用会话，您需要与启用会话的实体一起工作。

• 发送和接收会话消息
• 使用会话处理器

使用 ASP.NET Core 依赖注入进行注册

要在一个 ASP.NET Core 应用中将 ServiceBusClient 作为依赖项注入，安装 Azure 客户端库集成 ASP.NET Core 包。

dotnet add package Microsoft.Extensions.Azure

然后在配置您的服务的地方注册客户端。对于 ASP.NET Core 应用，这通常在 Program.cs 或 StartupConfigureServices 方法中直接进行。

public void ConfigureServices(IServiceCollection services)
{
    services.AddAzureClients(builder =>
    {
        builder.AddServiceBusClient("<< SERVICE BUS CONNECTION STRING >>");
    });

    // Register other services, controllers, and other infrastructure.
}

对于更喜欢使用共享 Azure.Identity 凭证的客户端的应用程序，注册看起来略有不同

public void ConfigureServices(IServiceCollection services)
 {
     services.AddAzureClients(builder =>
     {
         // This will register the ServiceBusClient using an Azure Identity credential.
         builder.AddServiceBusClientWithNamespace("<< YOUR NAMESPACE >>.servicebus.windows.net");

         // By default, DefaultAzureCredential is used, which is likely desired for most
         // scenarios. If you need to restrict to a specific credential instance, you could
         // register that instance as the default credential instead.
         builder.UseCredential(new ManagedIdentityCredential());
     });

     // Register other services, controllers, and other infrastructure.
 }

还可以使用已注册的 ServiceBusClient 实例通过依赖注入（DI）注册子客户端，例如 ServiceBusSender 和 ServiceBusReceiver。例如，为了注册属于命名空间的每个队列的发送者

public async Task ConfigureServicesAsync(IServiceCollection services)
{
    // Query the available queues for the Service Bus namespace.
    var adminClient = new ServiceBusAdministrationClient("<< SERVICE BUS CONNECTION STRING >>");
    var queueNames = new List<string>();

    // Because the result is async, they need to be captured to a standard list to avoid async
    // calls when registering.  Failure to do so results in an error with the services collection.
    await foreach (var queue in adminClient.GetQueuesAsync())
    {
        queueNames.Add(queue.Name);
    }

    // After registering the ServiceBusClient, register a named factory for each
    // queue.  This allows them to be lazily created and managed as singleton instances.

    services.AddAzureClients(builder =>
    {
        builder.AddServiceBusClient("<< SERVICE BUS CONNECTION STRING >>");

        foreach (var queueName in queueNames)
        {
            builder.AddClient<ServiceBusSender, ServiceBusClientOptions>((_, _, provider) =>
                provider
                    .GetService<ServiceBusClient>()
                    .CreateSender(queueName)
            )
            .WithName(queueName);
        }
    });

    // Register other services, controllers, and other infrastructure.
}

由于发送者是按其关联的队列命名的，因此注入时不必直接绑定到它们。相反，您将绑定到一个工厂，该工厂可以用于检索命名发送者

public class ServiceBusSendingController : ControllerBase
{
    private readonly ServiceBusSender _sender;

    public ServiceBusSendingController(IAzureClientFactory<ServiceBusSender> serviceBusSenderFactory)
    {
        // Though the method is called "CreateClient", the factory will manage the sender as a
        // singleton, creating a new instance only on the first use.
        _sender = serviceBusSenderFactory.CreateClient("<< QUEUE NAME >>");
    }
}

有关更多详细信息示例，请参阅Azure SDK for .NET 的依赖注入。

故障排除

请参阅Service Bus 故障排除指南。

下一步

除了介绍性场景之外，Azure Service Bus 客户端库还提供对其他场景的支持，以充分利用 Azure Service Bus 服务全部功能。为了帮助探索这些场景之一，Service Bus 客户端库提供了一个示例项目，用于说明常见场景。请参阅samples README获取详细信息。

贡献

该项目欢迎贡献和意见。大多数贡献需要在您同意贡献者许可协议（CLA），声明您有权实际上也是授予我们使用您贡献的权利。有关详细信息，请访问https://cla.microsoft.com。

当您提交拉取请求时， cla-bot 将自动确定您是否需要提供 CLA 并适当标记 PR（例如，标签，注释）。只需遵循机器人提供的说明即可。您只需在所有使用我们 CLA 的存储库中做一次。

该项目采用了Microsoft 开源行为准则。有关更多信息，请参阅行为准则常见问题解答或通过[email protected]与我们联系以提出任何其他问题或评论。

请参阅我们的贡献指南获取更多信息。