快速入门 - 向/从 Azure 服务总线队列 (.NET) 发送/接收消息

在本快速入门中,你将执行以下步骤:

  1. 使用 Azure 门户创建服务总线命名空间。
  2. 使用 Azure 门户创建服务总线队列。
  3. 编写 .NET 控制台应用程序,向队列发送一组消息。
  4. 编写 .NET 控制台应用程序,从队列接收这些消息。

注意

本快速入门分步介绍了如何实施一个简单方案,将一批消息发送到服务总线队列并接收这些消息。 有关 .NET 客户端库的概述,请参阅适用于 .NET 的 Azure 服务总线客户端库。 有关更多示例,请参阅 GitHub 上的服务总线 .NET 示例

先决条件

如果你是首次使用该服务,请在使用本快速入门之前先参阅服务总线概述

  • Azure 订阅。 若要使用 Azure 服务(包括 Azure 服务总线),需要一个订阅。 如果没有现有的 Azure 帐户,可以注册试用订阅
  • Visual Studio 2022。 示例应用程序利用 C# 10 中引入的新功能。 你仍可使用以前的 C# 语言版本的服务总线客户端库,但语法可能会有所不同。 若要使用最新语法,建议安装 .NET 6.0 或更高版本,并将语言版本设置为 latest。 如果使用 Visual Studio,Visual Studio 2022 以前的版本与生成 C# 10 项目时所需的工具将不兼容。

在 Azure 门户中创建命名空间

若要开始在 Azure 中使用服务总线消息实体,必须先使用在 Azure 中唯一的名称创建一个命名空间。 命名空间提供了用于应用程序中的服务总线资源的范围容器。

创建命名空间:

  1. 登录到 Azure 门户

  2. 在门户左侧的导航窗格中,选择“所有服务”,从类别列表中选择“集成”,将鼠标悬停在“服务总线”上,然后选择“服务总线”磁贴上的“创建”。

    图像显示“创建资源”、“集成”以及菜单中的“服务总线”选择。

  3. 在“创建命名空间”页的“基本信息”标记中,执行以下步骤 :

    1. 对于“订阅”,请选择要在其中创建命名空间的 Azure 订阅。

    2. 对于“资源组”,请选择该命名空间驻留到的现有资源组,或创建一个新资源组。

    3. 输入命名空间的名称。 命名空间名称应遵循以下命名约定:

      • 该名称在 Azure 中必须唯一。 系统会立即检查该名称是否可用。
      • 名称长度最少为 6 个字符,最多为 50 个字符。
      • 名称只能包含字母、数字和连字符“-”。
      • 名称必须以字母开头,并且必须以字母或数字结尾。
      • 名称不以“-sb”或“-mgmt”结尾。
    4. 对于“位置”,请选择托管该命名空间的区域。

    5. 对于“定价层”,请选择命名空间的定价层(“基本”、“标准”或“高级”)。 对于本快速入门,请选择“标准”。

      重要

      若要使用主题和订阅,请选择“标准”或“高级”。 基本定价层不支持主题/订阅。

      如果选择了“高级”定价层,请指定“消息传送单元”数 。 高级层在 CPU 和内存级别提供资源隔离,使每个工作负荷在隔离的环境中运行。 此资源容器称为消息传送单元。 高级命名空间至少具有一个消息传送单元。 可为每个服务总线高级命名空间选择 1、2、4、8 或 16 个消息传送单元。 有关详细信息,请参阅服务总线高级消息传送

    6. 在页面底部选择“查看 + 创建”。

      图像显示“创建命名空间”页

    7. 在“查看 + 创建”页上,查看设置,然后选择“创建” 。

  4. 资源部署成功后,在部署页上选择“转到资源”。

    图像显示“部署成功”页,其中包括“转到资源”链接。

  5. 将会看到服务总线命名空间的主页。

    图像显示已创建的服务总线命名空间的主页。

获取连接字符串

创建新的命名空间会自动生成一个初始共享访问签名 (SAS) 策略,该策略包含主密钥和辅助密钥以及主要连接字符串和辅助连接字符串,每个连接字符串都授予对命名空间所有方面的完全控制权。 请参阅服务总线身份验证和授权,了解如何创建规则来对普通发送者和接收者的权限进行更多限制。

客户端可以使用连接字符串连接到服务总线命名空间。 若要复制命名空间的主要连接字符串,请执行以下步骤:

  1. 在“服务总线命名空间”页中的左侧菜单上,选择“共享访问策略” 。

  2. 在“共享访问策略”页,选择“RootManageSharedAccessKey” 。

  3. 在“策略: RootManageSharedAccessKey”窗口中,选择“主连接字符串”旁边的“复制”按钮,将连接字符串复制到剪贴板供以后使用。 将此值粘贴到记事本或其他某个临时位置。

    屏幕截图显示了名为 RootManageSharedAccessKey 的 S A S 策略,其中包含密钥和连接字符串。

    可使用此页面复制主密钥、辅助密钥、主连接字符串和辅助连接字符串。

在 Azure 门户中创建队列

  1. 在“服务总线命名空间”页面上,选择左侧导航菜单中的“队列”

  2. 在“队列”页面上,选择工具栏上的“+ 队列”

  3. 输入队列名称,其他值则保留默认值。

  4. 现在选择“创建”。

    显示在门户中创建队列的图像

向 Azure 验证应用

对大多数 Azure 服务的应用程序请求必须获得授权。 要在代码中实现与 Azure 服务的无密码连接,推荐使用 Azure 标识客户端库提供的 DefaultAzureCredential 类。

还可以直接使用密码、连接字符串或其他凭据授权对 Azure 服务的请求。 但是,应谨慎使用此方法。 开发人员必须尽量避免在不安全的位置公开这些机密。 任何可获得密码或密钥的人员都可进行身份验证。 DefaultAzureCredential 提供比帐户密钥更好的管理和安全优势,来实现无密码身份验证。 以下示例演示了这两个选项。

DefaultAzureCredential 是适用于 .NET 的 Azure 标识客户端库提供的类。 有关 DefaultAzureCredential 的详细信息,请参阅 DefaultAzureCredential 概述DefaultAzureCredential 支持多种身份验证方法,并确定应在运行时使用哪种方法。 通过这种方法,你的应用可在不同环境(本地与生产)中使用不同的身份验证方法,而无需实现特定于环境的代码。

例如,应用可在本地开发时使用 Visual Studio 登录凭据进行身份验证,然后在将其部署到 Azure 后使用托管标识。 此转换不需要进行任何代码更改。

将角色分配给 Azure AD 用户

在本地开发时,请确保连接到 Azure 服务总线的用户帐户具有正确的权限。 你需要拥有 Azure 服务总线数据所有者角色才能发送和接收消息。 若要为自己分配此角色,需要具有“用户访问管理员”角色,或者具有包含 Microsoft.Authorization/roleAssignments/write 操作的其他角色。 可使用 Azure 门户、Azure CLI 或 Azure PowerShell 向用户分配 Azure RBAC 角色。 可在范围概述页上详细了解角色分配的可用范围。

以下示例将 Azure Service Bus Data Owner 角色分配给用户帐户,该角色提供对 Azure 服务总线资源的完全访问权限。 在实际方案中,遵循最小特权原则,仅向用户提供更安全的生产环境所需的最小权限。

适用于 Azure 服务总线的 Azure 内置角色

对于 Azure 服务总线,通过 Azure 门户和 Azure 资源管理 API 对命名空间和所有相关资源的管理已使用 Azure RBAC 模型进行了保护。 Azure 提供以下 Azure 内置角色,用于授予对服务总线命名空间的访问权限:

如果要创建自定义角色,请参阅执行服务总线操作所需的权限

重要

在大多数情况下,角色分配在 Azure 中传播需要一两分钟。 在极少数情况下,最多可能需要 8 分钟。 如果在首次运行代码时收到身份验证错误,请稍等片刻再试。

  1. 在 Azure 门户中,使用主搜索栏或左侧导航找到你的服务总线命名空间。

  2. 在概述页面上,从左侧菜单中选择“访问控制(IAM)”。

  3. 在“访问控制 (IAM)”页上,选择“角色分配”选项卡。

  4. 从顶部菜单中选择“+ 添加”,然后从出现的下拉菜单中选择“添加角色分配”。

    显示如何分配角色的屏幕截图。

  5. 使用搜索框将结果筛选为所需角色。 对于此示例,请搜索 Azure Service Bus Data Owner 并选择匹配的结果。 然后选择“下一步” 。

  6. 在“访问权限分配对象”下,选择“用户、组或服务主体”,然后选择“+ 选择成员”。

  7. 在对话框中,搜索 Azure AD 用户名(通常为 user@domain 电子邮件地址),然后选择对话框底部的“选择”。

  8. 选择“查看 + 分配”转到最后一页,然后再次选择“查看 + 分配”完成该过程。

登录并添加 Azure 标识包

可以使用以下步骤授权对服务总线命名空间的访问:

确保使用将角色分配到的同一 Azure AD 帐户进行身份验证。 可通过 Azure CLI、Visual Studio 或 Azure PowerShell 进行身份验证。

使用以下命令通过 Azure CLI 登录到 Azure:

az login

若要使用 DefaultAzureCredential,请将 Azure.Identity 包添加到应用程序。

  1. 在解决方案资源管理器中,右键单击项目的“依赖项”节点。 选择“管理 NuGet 包”。

  2. 在出现的窗口中,搜索 Azure.Identity。 选择相应的结果,然后选择“安装”。

    显示如何添加标识包的屏幕截图。

启动 Visual Studio 并登录到 Azure

可以使用以下步骤授权对服务总线命名空间的访问:

  1. 启动 Visual Studio。 如果看到“入门”窗口,请在右窗格中选择“在不使用代码的情况下继续”链接。

  2. 选择 Visual Studio 右上角的“登录”按钮。

    显示使用 Visual Studio 时用于登录到 Azure 的按钮的屏幕截图。

  3. 使用你之前为其分配角色的 Azure AD 帐户登录。

    显示帐户选择的屏幕截图。

将消息发送到队列

本部分介绍如何创建一个向服务总线队列发送消息的 .NET 控制台应用程序。

注意

本快速入门分步介绍了如何实施一个简单方案,将一批消息发送到服务总线队列并接收这些消息。 有关其他和高级方案的更多示例,请参阅 GitHub 上的服务总线 .NET 示例

创建控制台应用程序

  1. 在 Visual Studio 中,选择“文件”->“新建”->“项目”菜单。

  2. 在“创建新项目”对话框中执行以下步骤:如果看不到此对话框,请在菜单中选择“文件”,然后依次选择“新建”、“项目”。

    1. 选择“C#”作为编程语言。

    2. 选择“控制台”作为应用程序类型。

    3. 从结果列表中选择“控制台应用”。

    4. 然后,选择“下一步” 。

      显示使用 C# 和所选的控制台创建新项目对话框的图像

  3. 输入 QueueSender 作为项目名称,输入 ServiceBusQueueQuickStart 作为解决方案名称,然后选择下一步

    在“配置你的新项目对话框”中显示解决方案和项目名称的图像

  4. 在“其他信息”页面,选择“创建”来创建解决方案和项目。

向项目添加 NuGet 包

  1. 在菜单中选择“工具”>“NuGet 包管理器”>“包管理器控制台”

  2. 运行以下命令以安装 Azure.Messaging.ServiceBus NuGet 包。

    Install-Package Azure.Messaging.ServiceBus
    
  3. 运行以下命令以安装 Azure.Identity NuGet 包。

    Install-Package Azure.Identity
    

添加将消息发送到队列的代码

  1. Program.cs 的内容替换为以下代码。 下面概述了重要步骤,并在代码注释中提供了其他信息。

    重要

    使用服务总线命名空间和队列的名称更新代码片段中的占位符值(<NAMESPACE-CONNECTION-STRING><QUEUE-NAME>)。

    using Azure.Messaging.ServiceBus;
    using Azure.Identity;
    
    // name of your Service Bus queue
    // the client that owns the connection and can be used to create senders and receivers
    ServiceBusClient client;
    
    // the sender used to publish messages to the queue
    ServiceBusSender sender;
    
    // number of messages to be sent to the queue
    const int numOfMessages = 3;
    
    // The Service Bus client types are safe to cache and use as a singleton for the lifetime
    // of the application, which is best practice when messages are being published or read
    // regularly.
    //
    // Set the transport type to AmqpWebSockets so that the ServiceBusClient uses the port 443. 
    // If you use the default AmqpTcp, ensure that ports 5671 and 5672 are open.
    var clientOptions = new ServiceBusClientOptions
    { 
        TransportType = ServiceBusTransportType.AmqpWebSockets
    };
    //TODO: Replace the "<NAMESPACE-NAME>" and "<QUEUE-NAME>" placeholders.
    client = new ServiceBusClient(
        "<NAMESPACE-NAME>.servicebus.chinacloudapi.cn",
        new DefaultAzureCredential(),
        clientOptions);
    sender = client.CreateSender("<QUEUE-NAME>");
    
    // create a batch 
    using ServiceBusMessageBatch messageBatch = await sender.CreateMessageBatchAsync();
    
    for (int i = 1; i <= numOfMessages; i++)
    {
        // try adding a message to the batch
        if (!messageBatch.TryAddMessage(new ServiceBusMessage($"Message {i}")))
        {
            // if it is too large for the batch
            throw new Exception($"The message {i} is too large to fit in the batch.");
        }
    }
    
    try
    {
        // Use the producer client to send the batch of messages to the Service Bus queue
        await sender.SendMessagesAsync(messageBatch);
        Console.WriteLine($"A batch of {numOfMessages} messages has been published to the queue.");
    }
    finally
    {
        // Calling DisposeAsync on client types is required to ensure that network
        // resources and other unmanaged objects are properly cleaned up.
        await sender.DisposeAsync();
        await client.DisposeAsync();
    }
    
    Console.WriteLine("Press any key to end the application");
    Console.ReadKey();
    
  2. 生成项目并确保没有错误。

  3. 运行程序并等待出现确认消息。

    A batch of 3 messages has been published to the queue
    

    重要

    在大多数情况下,角色分配在 Azure 中传播需要一两分钟。 在极少数情况下,最多可能需要 8 分钟。 如果在首次运行代码时收到身份验证错误,请稍等片刻再试。

  4. 在 Azure 门户中按照以下步骤操作:

    1. 导航到服务总线命名空间。

    2. 在“概述”页上,在底部中间的窗格中选择队列。

      按照所选队列在 Azure 门户中显示服务总线命名空间页面的图像。

    3. 请注意“必备”部分中的值。

      显示接收的消息数量和队列大小的图像。

    请注意以下值:

    • 队列的活动消息计数值现在为 3。 每次运行此发件人应用而不检索消息时,该值会增加 3。
    • 每次应用向队列添加消息时,队列的当前大小都会递增。
    • 消息图表中的底部指标部分中,可以看到队列有三条传入的消息。

从队列接收消息

在本部分中,你将创建一个 .NET 控制台应用程序,用于接收来自队列的消息。

注意

本快速入门分步介绍了如何实施一个方案,将一批消息发送到服务总线队列并接收这些消息。 有关其他和高级方案的更多示例,请参阅 GitHub 上的服务总线 .NET 示例

为接收器创建项目

  1. 在“解决方案资源管理器”窗口中,右键单击 ServiceBusQueueQuickStart 解决方案,指向添加,然后选择新建项目
  2. 选择控制台应用程序,然后选择下一步
  3. 输入 QueueReceiver 作为项目名称,然后选择创建
  4. 解决方案资源管理器窗口中,右键单击 QueueReceiver,然后选择设为启动项目

向项目添加 NuGet 包

  1. 在菜单中选择“工具”>“NuGet 包管理器”>“包管理器控制台”

  2. 选择“QueueReceiver”作为“默认项目”。

    显示在“包管理器”控制台中选择了 QueueReceiver 项目的屏幕截图。

  3. 运行以下命令以安装 Azure.Messaging.ServiceBus NuGet 包。

    Install-Package Azure.Messaging.ServiceBus
    
  4. 运行以下命令以安装 Azure.Identity NuGet 包。

    Install-Package Azure.Identity
    

添加从队列接收消息的代码

在本部分中,你将添加从队列检索消息的代码。

  1. Program 类中,添加以下代码:

    using System.Threading.Tasks;
    using Azure.Identity;
    using Azure.Messaging.ServiceBus;
    
    // the client that owns the connection and can be used to create senders and receivers
    ServiceBusClient client;
    
    // the processor that reads and processes messages from the queue
    ServiceBusProcessor processor;
    
  2. 将以下方法追加到 Program 类的末尾。

    // handle received messages
    async Task MessageHandler(ProcessMessageEventArgs args)
    {
        string body = args.Message.Body.ToString();
        Console.WriteLine($"Received: {body}");
    
        // complete the message. message is deleted from the queue. 
        await args.CompleteMessageAsync(args.Message);
    }
    
    // handle any errors when receiving messages
    Task ErrorHandler(ProcessErrorEventArgs args)
    {
        Console.WriteLine(args.Exception.ToString());
        return Task.CompletedTask;
    }
    
  3. 将以下代码追加到 Program 类的末尾。 下面概述了重要步骤,并在代码注释中提供了其他信息。

    重要

    使用服务总线命名空间和队列的名称更新代码片段中的占位符值(<NAMESPACE-NAME><QUEUE-NAME>)。

    // The Service Bus client types are safe to cache and use as a singleton for the lifetime
    // of the application, which is best practice when messages are being published or read
    // regularly.
    //
    // Set the transport type to AmqpWebSockets so that the ServiceBusClient uses port 443. 
    // If you use the default AmqpTcp, make sure that ports 5671 and 5672 are open.
    
    // TODO: Replace the <NAMESPACE-NAME> placeholder
    var clientOptions = new ServiceBusClientOptions()
    {
        TransportType = ServiceBusTransportType.AmqpWebSockets
    };
    client = new ServiceBusClient(
        "<NAMESPACE-NAME>.servicebus.chinacloudapi.cn",
        new DefaultAzureCredential(),
        clientOptions);
    
    // create a processor that we can use to process the messages
    // TODO: Replace the <QUEUE-NAME> placeholder
    processor = client.CreateProcessor("<QUEUE-NAME>", new ServiceBusProcessorOptions());
    
    try
    {
        // add handler to process messages
        processor.ProcessMessageAsync += MessageHandler;
    
        // add handler to process any errors
        processor.ProcessErrorAsync += ErrorHandler;
    
        // start processing 
        await processor.StartProcessingAsync();
    
        Console.WriteLine("Wait for a minute and then press any key to end the processing");
        Console.ReadKey();
    
        // stop processing 
        Console.WriteLine("\nStopping the receiver...");
        await processor.StopProcessingAsync();
        Console.WriteLine("Stopped receiving messages");
    }
    finally
    {
        // Calling DisposeAsync on client types is required to ensure that network
        // resources and other unmanaged objects are properly cleaned up.
        await processor.DisposeAsync();
        await client.DisposeAsync();
    }
    
  4. 完成的 Program 类应与以下代码匹配:

    using System.Threading.Tasks;
    using Azure.Messaging.ServiceBus;
    using Azure.Identity;
    
    // the client that owns the connection and can be used to create senders and receivers
    ServiceBusClient client;
    
    // the processor that reads and processes messages from the queue
    ServiceBusProcessor processor;
    
    // The Service Bus client types are safe to cache and use as a singleton for the lifetime
    // of the application, which is best practice when messages are being published or read
    // regularly.
    //
    // Set the transport type to AmqpWebSockets so that the ServiceBusClient uses port 443.
    // If you use the default AmqpTcp, make sure that ports 5671 and 5672 are open.
    
    // TODO: Replace the <NAMESPACE-NAME> and <QUEUE-NAME> placeholders
    var clientOptions = new ServiceBusClientOptions() 
    {
        TransportType = ServiceBusTransportType.AmqpWebSockets
    };
    client = new ServiceBusClient("<NAMESPACE-NAME>.servicebus.chinacloudapi.cn", 
        new DefaultAzureCredential(), clientOptions);
    
    // create a processor that we can use to process the messages
    // TODO: Replace the <QUEUE-NAME> placeholder
    processor = client.CreateProcessor("<QUEUE-NAME>", new ServiceBusProcessorOptions());
    
    try
    {
        // add handler to process messages
        processor.ProcessMessageAsync += MessageHandler;
    
        // add handler to process any errors
        processor.ProcessErrorAsync += ErrorHandler;
    
        // start processing 
        await processor.StartProcessingAsync();
    
        Console.WriteLine("Wait for a minute and then press any key to end the processing");
        Console.ReadKey();
    
        // stop processing 
        Console.WriteLine("\nStopping the receiver...");
        await processor.StopProcessingAsync();
        Console.WriteLine("Stopped receiving messages");
    }
    finally
    {
        // Calling DisposeAsync on client types is required to ensure that network
        // resources and other unmanaged objects are properly cleaned up.
        await processor.DisposeAsync();
        await client.DisposeAsync();
    }
    
    // handle received messages
    async Task MessageHandler(ProcessMessageEventArgs args)
    {
        string body = args.Message.Body.ToString();
        Console.WriteLine($"Received: {body}");
    
        // complete the message. message is deleted from the queue. 
        await args.CompleteMessageAsync(args.Message);
    }
    
    // handle any errors when receiving messages
    Task ErrorHandler(ProcessErrorEventArgs args)
    {
        Console.WriteLine(args.Exception.ToString());
        return Task.CompletedTask;
    }
    
  5. 生成项目并确保没有错误。

  6. 运行接收器应用程序。 你应该会看到接收的消息。 按任意键来停止使用接收器和应用程序。

    Wait for a minute and then press any key to end the processing
    Received: Message 1
    Received: Message 2
    Received: Message 3
    
    Stopping the receiver...
    Stopped receiving messages
    
  7. 再次检查门户。 等待几分钟,如果未看到活动消息的 0,请刷新页面。

    • 活动消息计数和当前大小值现为 0

    • 在“消息”图表底部的“指标”部分中,可以看到队列有三条传入消息和三条传出消息 。

      显示接收后的活动消息和大小的屏幕截图。

清理资源

导航到 Azure 门户中的服务总线命名空间,然后在 Azure 门户上选择“删除”以删除命名空间及其中的队列。

另请参阅

请参阅以下文档和示例:

后续步骤