将消息发送到Azure Service Bus主题，并从主题订阅接收消息（Python）

将角色分配给Microsoft Entra用户

在本地开发时，请确保连接到Azure Service Bus的用户帐户具有正确的权限。 需要Azure Service Bus数据所有者角色才能发送和接收消息。 若要将此角色分配给自己，您需要 "User Access Administrator" 角色或其他包含 Microsoft.Authorization/roleAssignments/write 操作的角色。

可以使用Azure门户、Azure CLI或Azure PowerShell向用户分配Azure RBAC 角色。 若要详细了解角色分配的可用范围，请参阅 Azure RBAC 的可用范围。

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

Azure Service Bus 的 Azure 内置角色

对于Azure Service Bus，通过 Azure 门户和Azure资源管理 API 管理命名空间和所有相关资源已使用 Azure RBAC 模型进行保护。 Azure提供以下Azure内置角色，用于授权访问Service Bus命名空间：

• Azure Service Bus数据所有者：允许数据访问Service Bus命名空间及其实体，包括队列、主题、订阅和筛选器。 此角色的成员可以从队列或主题/订阅发送和接收消息。
• Azure Service Bus数据发送方：使用此角色授予对Service Bus命名空间及其实体send访问权限。
• Azure Service Bus数据接收器：使用此角色向receive授予对Service Bus命名空间及其实体的访问权限。

若要创建自定义角色，请参阅 Service Bus 操作所需的权限。

将Microsoft Entra用户添加到Azure Service Bus所有者角色

将您的 Microsoft Entra 用户名添加到 Service Bus 命名空间层级的 Azure Service Bus 数据所有者 角色。 此配置允许在用户帐户上下文中运行的应用将消息发送到队列或主题。 它可以接收来自队列或主题订阅的消息。

1. 如果没有在Azure门户中打开Service Bus命名空间页，请使用主搜索栏或左侧导航找到Service Bus命名空间。

2. 在 “概述 ”页上，从左侧菜单中选择 “访问控制”（IAM ）。

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

4. 从顶部菜单中选择 “+ 添加” ，然后 添加角色分配。

   

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

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

7. 在对话框中，搜索Microsoft Entra用户名（通常是user@domain电子邮件地址），然后选择对话框底部的选择。

8. 选择“审核 + 指派”转到最后一页，然后再次选择“审核 + 指派”以完成该过程。

获取连接字符串

创建新的命名空间会自动生成具有主密钥和辅助密钥的初始共享访问签名（SAS）策略。 它创建主连接字符串和辅助连接字符串，每个连接字符串都授予对命名空间的各个方面的完全控制。 有关如何为常规发送方和接收方创建具有更多约束权限的规则的详细信息，请参阅 Service Bus身份验证和授权。

客户端可以使用connection string连接到Service Bus命名空间。 若要复制命名空间的主连接字符串，请执行以下步骤：

1. 在 Service Bus Namespace 页面上，展开设置，随后选择“共享访问策略”。

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

3. 在 Policy： RootManageSharedAccessKey 窗口中，选择 Primary Connection String 旁边的复制按钮，将connection string复制到剪贴板供以后使用。 将此值粘贴到记事本或其他某个临时位置。

   

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

若要遵循本快速入门，请使用无密码身份验证以及您自己的 Azure 账户：

• 安装 Azure CLI。
• 请在终端或命令提示符下使用 az login 登录您的 Azure 帐户。
• 请在本教程后续部分将适当角色添加到资源时使用同一帐户。
• 在同样的终端或命令提示符中运行教程代码。

若要使用无密码代码，需要指定：

• 完全限定的服务总线命名空间，例如：<service-bus-namespace.servicebus.chinacloudapi.cn>
• 主题名称
• 订阅名称

若要按照本快速上手指南中使用连接字符串进行身份验证，请勿使用自己的 Azure 帐户。 而是使用服务总线命名空间的连接字符串。

若要使用连接代码，需要指定：

• 连接字符串
• 主题名称
• 订阅名称

1. 若要安装此Service Bus教程所需的Python包，请打开其路径中Python的命令提示符。 将目录更改为要在其中保存示例的文件夹。

2. 安装软件包：

   pip install azure-servicebus
   pip install azure-identity
   pip install aiohttp
   

1. 若要安装此Service Bus教程所需的Python包，请打开其路径中Python的命令提示符。 将目录更改为要在其中保存示例的文件夹。

2. 安装包：

   pip install azure-servicebus
   

1. 添加以下 import 语句。

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   from azure.servicebus import ServiceBusMessage
   from azure.identity.aio import DefaultAzureCredential
   

2. 添加常量并定义凭据。

   FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
   TOPIC_NAME = "TOPIC_NAME"
   
   credential = DefaultAzureCredential()
   

   重要

   • 将 FULLY_QUALIFIED_NAMESPACE 替换为该 Service Bus 命名空间的完全限定命名空间。
   • 将 TOPIC_NAME 替换为主题名称。

   在前面的代码中，你使用了 Azure Identity 客户端库的 DefaultAzureCredential 类。 当应用在开发期间在本地运行时，DefaultAzureCredential 将自动发现并使用您登录 Azure CLI 时使用的帐户进行身份验证。 将应用部署到Azure时，DefaultAzureCredential无需更改任何代码即可通过托管标识对应用进行身份验证以Microsoft Entra ID。

3. 添加一个方法以发送一条消息。

   async def send_single_message(sender):
       # Create a Service Bus message
       message = ServiceBusMessage("Single Message")
       # send the message to the topic
       await sender.send_messages(message)
       print("Sent a single message")
   

   发送方是一个对象，充当你创建的主题的客户端。 稍后将创建它，并将其作为参数发送到此函数。

4. 添加一个方法以发送一列消息。

   async def send_a_list_of_messages(sender):
       # Create a list of messages
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       # send the list of messages to the topic
       await sender.send_messages(messages)
       print("Sent a list of 5 messages")
   

5. 添加一个方法以发送一批消息。

   async def send_batch_message(sender):
       # Create a batch of messages
       async with sender:
           batch_message = await sender.create_message_batch()
           for _ in range(10):
               try:
                   # Add a message to the batch
                   batch_message.add_message(ServiceBusMessage("Message inside a ServiceBusMessageBatch"))
               except ValueError:
                   # ServiceBusMessageBatch object reaches max_size.
                   # New ServiceBusMessageBatch object can be created here to send more data.
                   break
           # Send the batch of messages to the topic
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. 创建一个Service Bus客户端，然后创建一个用于发送消息的主题发送者对象。

   async def run():
       # create a Service Bus client using the credential.
       async with ServiceBusClient(
           fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
           credential=credential,
           logging_enable=True) as servicebus_client:
           # Get a Topic Sender object to send messages to the topic
           sender = servicebus_client.get_topic_sender(topic_name=TOPIC_NAME)
           async with sender:
               # Send one message
               await send_single_message(sender)
               # Send a list of messages
               await send_a_list_of_messages(sender)
               # Send a batch of messages
               await send_batch_message(sender)
           # Close credential when no longer needed.
           await credential.close()
   
   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. 添加以下 import 语句。

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   from azure.servicebus import ServiceBusMessage
   

2. 添加以下常量。

   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STRING"
   TOPIC_NAME = "TOPIC_NAME"
   

   重要

   • 将NAMESPACE_CONNECTION_STRING替换为命名空间的连接字符串。
   • 将 TOPIC_NAME 替换为主题名称。

3. 添加一个方法以发送一条消息。

   async def send_single_message(sender):
       # Create a Service Bus message
       message = ServiceBusMessage("Single Message")
       # send the message to the topic
       await sender.send_messages(message)
       print("Sent a single message")
   

   发送方是一个对象，充当你创建的主题的客户端。 稍后将创建它，并将其作为参数发送到此函数。

4. 添加一个方法以发送一列消息。

   async def send_a_list_of_messages(sender):
       # Create a list of messages
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       # send the list of messages to the topic
       await sender.send_messages(messages)
       print("Sent a list of 5 messages")
   

5. 添加一个方法以发送一批消息。

   async def send_batch_message(sender):
       # Create a batch of messages
       async with sender:
           batch_message = await sender.create_message_batch()
           for _ in range(10):
               try:
                   # Add a message to the batch
                   batch_message.add_message(ServiceBusMessage("Message inside a ServiceBusMessageBatch"))
               except ValueError:
                   # ServiceBusMessageBatch object reaches max_size.
                   # New ServiceBusMessageBatch object can be created here to send more data.
                   break
           # Send the batch of messages to the topic
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. 创建一个Service Bus客户端，然后创建一个用于发送消息的主题发送者对象。

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient.from_connection_string(
           conn_str=NAMESPACE_CONNECTION_STR,
           logging_enable=True) as servicebus_client:
           # Get a Topic Sender object to send messages to the topic
           sender = servicebus_client.get_topic_sender(topic_name=TOPIC_NAME)
           async with sender:
               # Send one message
               await send_single_message(sender)
               # Send a list of messages
               await send_a_list_of_messages(sender)
               # Send a batch of messages
               await send_batch_message(sender)
   
   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. 与 send 示例类似，添加 import 语句，定义应替换为自己的值的常量，并定义凭据。

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   from azure.identity.aio import DefaultAzureCredential
   
   FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
   SUBSCRIPTION_NAME = "SUBSCRIPTION_NAME"
   TOPIC_NAME = "TOPIC_NAME"
   
   credential = DefaultAzureCredential()
   

2. 创建Service Bus客户端，然后创建用于接收消息的订阅接收方对象。

   async def run():
       # create a Service Bus client using the credential
       async with ServiceBusClient(
           fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
           credential=credential,
           logging_enable=True) as servicebus_client:
   
           async with servicebus_client:
               # get the Subscription Receiver object for the subscription
               receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_NAME, 
               subscription_name=SUBSCRIPTION_NAME, max_wait_time=5)
               async with receiver:
                   received_msgs = await receiver.receive_messages(max_wait_time=5, max_message_count=20)
                   for msg in received_msgs:
                       print("Received: " + str(msg))
                       # complete the message so that the message is removed from the subscription
                       await receiver.complete_message(msg)
           # Close credential when no longer needed.
           await credential.close()
   

3. 调用 run 方法。

   asyncio.run(run())
   

1. 与 send 示例类似，添加 import 语句并定义应替换为自己的值的常量。

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   
   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STRING"
   SUBSCRIPTION_NAME = "SUBSCRIPTION_NAME"
   TOPIC_NAME = "TOPIC_NAME"
   

2. 创建Service Bus客户端，然后创建用于接收消息的订阅接收方对象。

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient.from_connection_string(
           conn_str=NAMESPACE_CONNECTION_STR,
           logging_enable=True) as servicebus_client:
   
           async with servicebus_client:
               # get the Subscription Receiver object for the subscription
               receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_NAME, 
               subscription_name=SUBSCRIPTION_NAME, max_wait_time=5)
               async with receiver:
                   received_msgs = await receiver.receive_messages(max_wait_time=5, max_message_count=20)
                   for msg in received_msgs:
                       print("Received: " + str(msg))
                       # complete the message so that the message is removed from the subscription
                       await receiver.complete_message(msg)
   

3. 调用 run 方法。

   asyncio.run(run())