你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

向 Azure 服务总线主题发送消息，并从该主题的订阅接收消息 (Python)

项目
02/06/2024

在本教程中，将完成以下步骤：

使用 Azure 门户创建服务总线命名空间。
使用 Azure 门户创建服务总线主题。
使用 Azure 门户创建该主题的服务总线订阅。
编写 Python 应用程序，使用 azure-servicebus 包执行以下操作：
- 将一组消息发送到主题。
- 从订阅接收这些消息。

注意

本快速入门分步说明了将一批消息发送到某个服务总线主题并从该主题的订阅接收这些消息这一简单场景。可在 GitHub 上的 Azure SDK for Python 存储库中找到预生成的 Azure 服务总线 Python 示例。

先决条件

一个 Azure 订阅。
Python 3.8 或更高版本

注意

本教程将演练可以使用 Python 复制和运行的示例。有关如何创建 Python 应用程序的说明，请参阅创建 Python 应用程序并将其部署到 Azure 网站。有关安装本教程中使用的包的详细信息，请参阅 Python 安装指南。

在 Azure 门户中创建命名空间

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

创建命名空间：

登录 Azure 门户。
导航到“所有服务”页。
在左侧的导航栏中，从类别列表中选择“集成”，将鼠标悬停在“服务总线”上，然后选择“服务总线”磁贴上的 + 按钮。。
在“创建命名空间”页的“基本信息”标记中，执行以下步骤：
1. 对于“订阅”，请选择要在其中创建命名空间的 Azure 订阅。
2. 对于“资源组”，请选择该命名空间驻留到的现有资源组，或创建一个新资源组。
3. 输入命名空间的名称。命名空间名称应遵循以下命名约定：
  - 该名称在 Azure 中必须唯一。系统会立即检查该名称是否可用。
  - 名称长度最少为 6 个字符，最多为 50 个字符。
  - 名称只能包含字母、数字、连字符“-”。
  - 名称必须以字母开头，并以字母或数字结尾。
  - 名称不以“-sb”或“-mgmt”结尾。
4. 对于“位置”，请选择托管该命名空间的区域。
5. 对于“定价层”，请选择命名空间的定价层（“基本”、“标准”或“高级”）。对于本快速入门，请选择“标准”。
  
  重要
  
  若要使用主题和订阅，请选择“标准”或“高级”。基本定价层不支持主题/订阅。
  
  如果选择了“高级”定价层，请指定“消息传送单元”数。高级层在 CPU 和内存级别提供资源隔离，使每个工作负荷在隔离的环境中运行。此资源容器称为消息传送单元。高级命名空间至少具有一个消息传送单元。可为每个服务总线高级命名空间选择 1、2、4、8 或 16 个消息传送单元。有关详细信息，请参阅服务总线高级消息传送。
6. 在页面底部选择“查看 + 创建”。
7. 在“查看 + 创建”页上，查看设置，然后选择“创建” 。
资源部署成功后，在部署页上选择“转到资源”。
将会看到服务总线命名空间的主页。

使用 Azure 门户创建主题

在“服务总线命名空间”页面上，选择左侧菜单中的“主题”。
在工具栏中选择“+ 主题”。
输入主题名称。将其他选项保留默认值。
选择“创建”。

创建主题的订阅

选择在上一部分创建的主题。
在“服务总线主题”页面上，选择工具栏上的“+ 订阅” 。
在“创建订阅”页上执行以下步骤：
1. 对于订阅名称，输入“S1” 。
2. 对于“最大交付数”，输入“3” 。
3. 然后，选择“创建”以创建订阅。

向 Azure 验证应用

本快速入门介绍了连接到 Azure 服务总线的两种方法：无密码和连接字符串方法。

第一个选项展示如何使用 Microsoft Entra ID 中的安全主体和基于角色的访问控制 (RBAC) 连接到服务总线命名空间。无需担心代码、配置文件或安全存储（如 Azure Key Vault）中存在硬编码的连接字符串。

第二个选项展示如何使用连接字符串连接到服务总线命名空间。如果不熟悉 Azure，你可能会感觉连接字符串选项更易于使用。建议在实际应用程序和生产环境中使用无密码选项。有关详细信息，请参阅身份验证和授权。还可以在概述页上阅读有关无密码身份验证的详细信息。

无密码（推荐）
连接字符串

将角色分配到 Microsoft Entra 用户

在本地开发时，请确保连接到 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 服务总线数据所有者：允许对服务总线命名空间及其实体（队列、主题、订阅和筛选器）进行数据访问。此角色的成员可以从队列或主题/订阅发送和接收消息。
Azure 服务总线数据发送者：使用此角色可以为服务总线命名空间及其实体提供发送访问权限。
Azure 服务总线数据接收者：使用此角色可以为服务总线命名空间及其实体提供接收访问权限。

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

将 Microsoft Entra 用户添加到“Azure 服务总线所有者”角色

将 Microsoft Entra 用户名添加到服务总线命名空间级别的 Azure 服务总线数据所有者角色。它将允许在用户帐户上下文中运行的应用将消息发送到队列或主题，并从队列或主题的订阅中接收消息。

重要

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

如果未在 Azure 门户中打开“服务总线命名空间”页，请使用主搜索栏或左侧导航找到你的服务总线命名空间。
在概述页面上，从左侧菜单中选择“访问控制(IAM)”。
在“访问控制 (IAM)”页上，选择“角色分配”选项卡。
从顶部菜单中选择“+ 添加”，然后从出现的下拉菜单中选择“添加角色分配”。
使用搜索框将结果筛选为所需角色。对于此示例，请搜索 Azure Service Bus Data Owner 并选择匹配的结果。然后选择“下一步” 。
在“访问权限分配对象”下，选择“用户、组或服务主体”，然后选择“+ 选择成员”。
在对话框中，搜索 Microsoft Entra ID 用户名（通常是 user@domain 电子邮件地址），然后选中对话框底部的“选择”。
选择“查看 + 分配”转到最后一页，然后再次选择“查看 + 分配”完成该过程。

若要使用无密码身份验证和你自己的 Azure 帐户来完成本快速入门，请执行以下操作：

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

重要

请确保使用 az login 进行登录。无密码代码中的 DefaultAzureCredential 类使用 Azure CLI 凭据通过 Microsoft Entra ID 进行身份验证。

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

完全限定的服务总线命名空间，例如 <service-bus-namespace>.servicebus.windows.net
主题名称
subscription name

使用 pip 安装包

无密码（推荐）
连接字符串

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

安装包：

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

若要安装此服务总线教程所需的 Python 包，请打开路径中包含 Python 的命令提示符。将目录更改为要在其中保存示例的文件夹。
安装包：
```
pip install azure-servicebus
```

将消息发送到主题

下面的示例代码演示如何将一批消息发送到服务总线主题。请参阅代码注释以了解详细信息。

打开自己偏好的编辑器（如 Visual Studio Code），创建 send.py 文件，并在其中添加以下代码。

无密码（推荐）
连接字符串

添加以下 import 语句。

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

添加常量并定义凭据。
```
FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
TOPIC_NAME = "TOPIC_NAME"

credential = DefaultAzureCredential()
```
重要
- 将 FULLY_QUALIFIED_NAMESPACE 替换为服务总线命名空间的完全限定命名空间。
- 将 TOPIC_NAME 替换为主题名称。
在前面的代码中，你使用了 Azure 标识客户端库的 DefaultAzureCredential 类。在开发期间，当应用在本地运行时，DefaultAzureCredential 将使用你登录到 Azure CLI 时使用的帐户自动发现 Azure 并向其进行身份验证。当应用部署到 Azure 时，DefaultAzureCredential 可以通过托管标识向 Microsoft Entra ID 验证你的应用，而无需更改任何代码。

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

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

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

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

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

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

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

创建一个服务总线客户端，然后创建一个主题发送方对象来发送消息。

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("-----------------------")

添加以下 import 语句。

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

添加以下常量。
```
NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STRING"
TOPIC_NAME = "TOPIC_NAME"
```
重要
- 将 NAMESPACE_CONNECTION_STRING 替换为命名空间的连接字符串。
- 将 TOPIC_NAME 替换为主题名称。

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

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

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

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

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

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

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

创建一个服务总线客户端，然后创建一个主题发送方对象来发送消息。

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("-----------------------")

从订阅接收消息

以下示例代码演示如何从订阅接收消息。此代码将持续接收新消息，直到在 5 (max_wait_time) 秒内未收到任何新消息。

打开自己偏好的编辑器（如 Visual Studio Code），创建 recv.py 文件，并在其中添加以下代码。

无密码（推荐）
连接字符串

与 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()

创建一个服务总线客户端，然后创建一个订阅接收方对象来接收消息。

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

调用 run 方法。
```
asyncio.run(run())
```

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

import asyncio
from azure.servicebus.aio import ServiceBusClient

NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STRING"
SUBSCRIPTION_NAME = "SUBSCRIPTION_NAME"
TOPIC_NAME = "TOPIC_NAME"

创建一个服务总线客户端，然后创建一个订阅接收方对象来接收消息。

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)

调用 run 方法。
```
asyncio.run(run())
```

运行应用

打开其路径中包含 Python 的命令提示符，然后运行代码以发送和接收某个主题下的订阅消息。

python send.py; python recv.py

应会看到以下输出：

Sent a single message
Sent a list of 5 messages
Sent a batch of 10 messages
Done sending messages
-----------------------
Received: Single Message
Received: Message in list
Received: Message in list
Received: Message in list
Received: Message in list
Received: Message in list
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch

在 Azure 门户中，导航到你的服务总线命名空间。在“概述”页上，验证传入和传出消息计数是否为 16 。如果没有看到这些消息，请等待几分钟后再刷新页面。

传入和传出消息计数

在底部窗格中选择主题，以查看主题的“服务总线主题”页。在此页上，应会在“消息”图表中看到三条传入消息和三条传出消息。

传入和传出消息

在此页上，如果选择一个订阅，则将转到“服务总线订阅”页。可以在此页上查看活动消息计数、死信消息计数等。在此示例中，所有消息均已接收，因此活动消息计数为零。

活动消息计数

如果注释掉接收代码，则会看到活动消息计数为 16。

活动消息计数 - 无接收

后续步骤

请参阅以下文档和示例：

适用于 Python 的 Azure 服务总线客户端库
示例。
- sync_samples 文件夹包含一些示例，这些示例演示如何以同步方式与服务总线交互。本快速入门就使用了此方法。
- async_samples 文件夹包含一些示例，这些示例演示如何以异步方式与服务总线交互。
azure-servicebus 参考文档

通过