python - 持续接收来自Azure ServiceBus的消息

Question

根据 Azure ServiceBus 文档 here :

The ServiceBusReceiver class defines a high level interface for receiving messages from the Azure Service Bus Queue or Topic Subscription. The two primary channels for message receipt are receive() to make a single request for messages, and async for message in receiver: to continuously receive incoming messages in an ongoing fashion.

我一直在尝试使用接收器中的消息异步:建议在每次出现消息时触发一个函数，但我不确定如何正确执行，因为我经验很少使用异步函数。熟悉异步/服务总线的人可以解释一下代码应该如何格式化吗？

编辑:让我提供更多背景信息。我正在创建一个 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())

完整教程: Send messages to an Azure Service Bus topic and receive messages from subscriptions to the topic (Python)

此外，您还可以探索这些示例(同步和异步版本):Azure Service Bus client library for Python Samples