Краткое руководство: Отправка и получение сообщений из очередей сервисной шины Azure (Python)

Чтобы использовать это краткое руководство с собственной учетной записью Azure, выполните указанные ниже действия.

• Установите Azure CLI, которая обеспечивает проверку подлинности без пароля на компьютере разработчика.
• Войдите с помощью учетной записи Azure в терминале или командной строке az login.
• Используйте ту же учетную запись при добавлении соответствующей роли данных в ресурс.
• Запустите код в том же терминале или командной строке.
• Запишите имя очереди для пространства имен служебная шина. Это необходимо в коде.

Обратите внимание на следующие значения, которые используются в коде:

• строка подключения пространства имен служебная шина
• созданная очередь пространства имен служебная шина

Назначение ролей пользователю Microsoft Entra

При локальной разработке убедитесь, что учетная запись пользователя, которая подключается к служебной шине Azure, имеет правильные разрешения. Для отправки и получения сообщений требуется роль владельца данных служебной шины Azure . Чтобы назначить себе эту роль, вам потребуется роль администратора доступа пользователей или другая роль, которая включает Microsoft.Authorization/roleAssignments/write действие.

Роли Azure RBAC можно назначить пользователю с помощью портала Azure, Azure CLI или Azure PowerShell. Чтобы узнать больше о доступных областях для назначения ролей, см. статью "Общие сведения об области применения Azure RBAC".

В следующем примере роль назначается Azure Service Bus Data Owner учетной записи пользователя, которая предоставляет полный доступ к ресурсам Служебная шина Azure. В реальном сценарии следуйте принципу наименьших привилегий , чтобы предоставить пользователям только минимальные разрешения, необходимые для более безопасной рабочей среды.

Встроенные роли Azure для служебной шины Azure

Для служебной шины Azure управление пространствами имен и всеми связанными ресурсами через портал Azure и API управления ресурсами Azure уже защищено с помощью модели Azure RBAC. Azure предоставляет следующие встроенные роли Azure для авторизации доступа к пространству имен служебная шина:

• Владелец данных служебной шины Azure: обеспечивает доступ к пространству имен служебной шины и его сущностям, включая очереди, разделы, подписки и фильтры. Участник этой роли может отправлять и получать сообщения из очередей или разделов или подписок.
• Отправитель данных служебной шины Azure: используйте эту роль для предоставления доступа к пространству send имен служебной шины и его сущностям.
• Приемник данных служебной шины Azure. Используйте эту роль, чтобы предоставить receive доступ к пространству имен служебной шины и его сущностям.

Если вы хотите создать пользовательскую роль, см. раздел "Права", необходимые для операций служебная шина.

Добавление пользователя Microsoft Entra в роль владельца Служебная шина Azure

Добавьте имя пользователя Microsoft Entra в роль владельца данных Служебная шина Azure на уровне пространства имен служебная шина. Эта конфигурация позволяет приложению, работающему в контексте учетной записи пользователя, отправлять сообщения в очередь или раздел. Он может получать сообщения из очереди или подписки на тему.

1. Если в портал Azure нет страницы пространства имен служебная шина, найдите пространство имен служебная шина с помощью главной панели поиска или навигации слева.

2. На странице обзора выберите элемент управления доступом (IAM) в меню слева.

3. На странице Контроль доступа (IAM) откройте вкладку Назначения ролей.

4. Выберите +Добавить из верхнего меню и добавьте назначение ролей.

   

5. Используйте поле поиска, чтобы отфильтровать результаты для отображения нужной роли. В этом примере найдите Azure Service Bus Data Owner и выберите соответствующий результат. Теперь щелкните Далее.

6. В разделе Назначение доступа для выберите Пользователь, группа или субъект-служба и + Выбрать членов.

7. В диалоговом окне найдите имя пользователя Microsoft Entra (обычно ваш user@domain адрес электронной почты), а затем выберите в нижней части диалогового окна.

8. Нажмите кнопку Проверить и назначить, чтобы перейти на последнюю страницу, а затем еще раз Проверить и назначить, чтобы завершить процесс.

Получение строки подключения

Создание нового пространства имен автоматически создает начальную политику для подписи общего доступа (SAS) с первичными и вторичными ключами. Он создает первичные и вторичные строки подключения, которые предоставляют полный контроль над всеми аспектами пространства имен. Для получения дополнительной информации о создании правил с более ограниченными правами для обычных отправителей и получателей, см. раздел аутентификация и авторизация Service Bus.

Клиент может использовать строка подключения для подключения к пространству имен служебная шина. Чтобы скопировать первичную строку подключения для пространства имен, выполните следующие действия:

1. На странице Service Bus Namespace в меню слева разверните Параметры, а затем выберите политики общего доступа.

2. На странице Политики общего доступа щелкните RootManageSharedAccessKey.

3. В окне Политика: RootManageSharedAccessKey нажмите кнопку копирования рядом с полем Первичная строка подключения, чтобы скопировать строку подключения в буфер обмена для последующего использования. Вставьте на время эти значения в Блокноте или любом другом месте.

   

   Эту страницу можно использовать для копирования первичного и вторичного ключей, а также первичной и вторичной строки подключения.

1. Чтобы установить необходимые пакеты Python для этого краткого руководства по служебной шине, откройте окно командной строки с python в пути.

2. Измените каталог на папку, в которой вы хотите получить примеры.

3. Установите следующие пакеты:

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

1. Чтобы установить необходимые пакеты Python для этого краткого руководства по служебной шине, откройте окно командной строки с python в пути.

2. Измените каталог на папку, в которой вы хотите получить примеры.

3. Установите следующий пакет:

   pip install azure-servicebus
   

1. Добавьте инструкции импорта.

   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"
   QUEUE_NAME = "QUEUE_NAME"
   
   credential = DefaultAzureCredential()
   

   Внимание

   • Замените FULLY_QUALIFIED_NAMESPACE полное пространство имен для пространства имен служебная шина.
   • Замените QUEUE_NAME именем очереди.

3. Добавьте метод для отправки одного сообщения.

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

   Отправитель — это объект, который выступает в качестве клиента для созданной очереди. Вы создадите его позже и отправите в качестве аргумента этой функции.

4. Добавьте метод для отправки списка сообщений.

   async def send_a_list_of_messages(sender):
       # Create a list of messages and send it to the queue
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       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 queue
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. Создайте клиента служебной шины, а затем объект отправителя очереди для отправки сообщений.

   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 Queue Sender object to send messages to the queue
           sender = servicebus_client.get_queue_sender(queue_name=QUEUE_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()
   

7. run Вызовите метод и распечатайте сообщение.

   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. Добавьте инструкции импорта.

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

2. Добавьте константы.

   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STR"
   QUEUE_NAME = "QUEUE_NAME"
   

   Внимание

   • Замените NAMESPACE_CONNECTION_STR строкой подключения для пространства имен Служебной шины.
   • Замените QUEUE_NAME именем очереди.

3. Добавьте метод для отправки одного сообщения.

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

   Отправитель — это объект, который выступает в качестве клиента для созданной очереди. Вы создадите его позже и отправите в качестве аргумента этой функции.

4. Добавьте метод для отправки списка сообщений.

   async def send_a_list_of_messages(sender):
       # Create a list of messages and send it to the queue
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       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 queue
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. Создайте клиента служебной шины, а затем объект отправителя очереди для отправки сообщений.

   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 Queue Sender object to send messages to the queue
           sender = servicebus_client.get_queue_sender(queue_name=QUEUE_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)
   

7. run Вызовите метод и распечатайте сообщение.

   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. Аналогично примеру с send.py, добавьте операторы import. Замените константы собственными значениями и определите учетные данные.

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

2. Создайте клиент служебная шина, а затем объект приемника очереди для получения сообщений.

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient(
           fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
           credential=credential,
           logging_enable=True) as servicebus_client:
   
           async with servicebus_client:
               # get the Queue Receiver object for the queue
               receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
               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 queue
                       await receiver.complete_message(msg)
   
           # Close credential when no longer needed.
           await credential.close()
   

3. Вызовите метод run .

   asyncio.run(run())
   

1. Как и в примере send.py , добавьте import инструкции и определите константы, использующие собственные значения.

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   
   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STR"
   QUEUE_NAME = "QUEUE_NAME"
   

2. Создайте клиент служебная шина, а затем объект приемника очереди для получения сообщений.

   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 Queue Receiver object for the queue
               receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
               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 queue
                       await receiver.complete_message(msg)
   

3. Вызовите метод run .

   asyncio.run(run())