持续接收来自Azure ServiceBus的消息

Question

根据此处的Azure ServiceBus 文档：

ServiceBusReceiver 类定义用于从 Azure 服务总线队列或主题订阅接收消息的高级接口。消息接收的两个主要通道是 receive()，用于发出单个消息请求，以及async for message in receiver:以持续的方式连续接收传入消息。

我一直在尝试使用async for message in receiver:建议在每次出现消息时触发函数，但我不确定如何正确执行，因为我几乎没有使用异步函数的经验。熟悉异步/服务总线的人可以解释一下代码应该如何格式化吗？

编辑：让我提供更多背景信息。我正在创建一个 python Flask 服务，在启动时，我需要它开始监听主题/订阅名称上的消息。每当它收到消息时，它都会执行一些代码，然后发回消息。那么...如何在启动时启动异步侦听器，并让它在触发时执行一些代码？它还应该能够以非阻塞的方式处理每条消息。因此，如果同时收到两条消息，则应同时处理两条消息。

注意：我无法使用 Azure Functions。

Answer 1

假设您正在使用topic-subscription，您可以使用以下代码：

#!/usr/bin/env python

# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

"""
Example to show receiving batch messages from a Service Bus Subscription under specific Topic asynchronously.
"""

# pylint: disable=C0111

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

CONNECTION_STR = os.environ['SERVICE_BUS_CONNECTION_STR']
TOPIC_NAME = os.environ["SERVICE_BUS_TOPIC_NAME"]
SUBSCRIPTION_NAME = os.environ["SERVICE_BUS_SUBSCRIPTION_NAME"]


async def main():
    servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR)

    async with servicebus_client:
        receiver = servicebus_client.get_subscription_receiver(
            topic_name=TOPIC_NAME,
            subscription_name=SUBSCRIPTION_NAME
        )
        async with receiver:
            received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5)
            for msg in received_msgs:
                print(str(msg))
                await receiver.complete_message(msg)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

完整教程： 向 Azure 服务总线主题发送消息并从该主题的订阅接收消息 (Python)

此外，您可以探索这些示例（同步和异步版本）：Python 示例的 Azure 服务总线客户端库