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

• C#
• Java
• JavaScript
• Python

В этом руководстве выполняются следующие шаги:

1. Создание пространства имен служебной шины с помощью портала Azure.
2. Создание очереди служебной шины с помощью портала Azure.
3. Написание кода Python для использования пакета azure-servicebus для:
   1. Отправьте набор сообщений в очередь.
   2. Получение этих сообщений из очереди.

Примечание.

В этом кратком руководстве содержатся пошаговые инструкции для простого сценария отправки сообщений в очередь Служебной шины и получения сообщений из них. Готовые примеры JavaScript и TypeScript для Служебной шины Azure находятся в репозитории пакета SDK Azure для Python на GitHub.

Необходимые компоненты

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

• Подписка Azure. Для работы с этим учебником требуется учетная запись Azure. Вы можете активировать преимущества подписчика MSDN или зарегистрироваться для получения бесплатной учетной записи.

• Python 3.8 или более поздней версии.

Без пароля (рекомендуется)

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

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

Строка подключения

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

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

Примечание.

В этом руководстве описаны примеры, которые можно скопировать и запустить с помощью Python. Инструкции по созданию приложения Python см. в статье "Создание и развертывание приложения Python на веб-сайте Azure". Дополнительные сведения об установке пакетов, используемых в этом руководстве, см. в руководстве по установке Python.

Создание пространства имен на портале Azure

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

Создание пространства имен службы:

1. Войдите на портал Azure.

2. Перейдите на страницу "Все службы".

3. На панели навигации слева выберите "Интеграция" из списка категорий, наведите указатель мыши на служебная шина, а затем нажмите + кнопку на плитке служебная шина.

   

4. В теге Основные сведения на странице Создание пространства имен выполните следующие действия:

   1. Выберите подписку Azure, в которой будет создано пространство имен.

   2. Выберите существующую группу ресурсов, в которую будет включено это пространство имен, или создайте новую.

   3. Введите имя для пространства имен. В имени пространства имен должны соблюдаться следующие соглашения об именовании:

      • Это имя должно быть уникальным в пределах Azure. Система немедленно проверяет, доступно ли оно.
      • Длина имени составляет не менее 6 и не более 50 символов.
      • Имя может содержать только буквы, цифры и дефисы (-).
      • Имя должно начинаться с буквы или цифры и заканчиваться буквой или цифрой.
      • Имя не должно оканчиваться на -sb или -mgmt.

   4. Укажите расположение — регион для размещения пространства имен.

   5. Для параметра Ценовая категория выберите ценовую категорию ("Базовый", "Стандартный" или "Премиум") для пространства имен. Для работы с этим кратким руководством выберите вариант Стандартный.

      Внимание

      Чтобы использовать разделы и подписки, выберите категорию "Стандартный" или "Премиум". Разделы и подписки не поддерживаются в ценовой категории "Базовый".

      Если выбрана ценовая категория Премиум, укажите число единиц обмена сообщениями. В категории "Премиум" обеспечивается изоляция ресурсов на уровне ЦП и памяти, так что рабочая нагрузка выполняется изолированно от других. Контейнер ресурса называется единицей обмена сообщениями. Пространству имен ценовой категории "Премиум" выделяется по крайней мере одна единица обмена сообщениями. Для каждого пространства имен служебной шины Premium можно выбрать 1, 2, 4, 8 или 16 единиц обмена сообщениями. Дополнительные сведения см. в статье Уровни обмена сообщениями через служебную шину Premium и Standard.

   6. В нижней части страницы выберите Review + create (Проверить и создать).

      

   7. На странице Проверить и создать проверьте параметры и нажмите кнопку Создать.

5. После успешного развертывания ресурса выберите "Перейти к ресурсу " на странице развертывания.

   

6. Вы увидите домашнюю страницу пространства имен служебной шины.

   

Создание очереди на портале Azure

1. На странице Пространство имен служебной шины в меню навигации слева выберите Очереди.

2. На странице Очереди на панели инструментов выберите + Очередь.

3. Введите имя очереди, остальные значения по умолчанию не изменяйте.

4. Выберите Создать.

   

Проверка подлинности приложения в Azure

В этом кратком руководстве показано два способа подключения к Служебная шина Azure: без пароля и строка подключения.

Первый вариант показывает, как использовать субъект безопасности в идентификаторе Microsoft Entra ID и управлении доступом на основе ролей (RBAC) для подключения к пространству имен служебная шина. Вам не нужно беспокоиться о наличии жестко закодированных строка подключения в коде или в файле конфигурации или в безопасном хранилище, например Azure Key Vault.

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

Без пароля (рекомендуется)

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

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

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

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

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

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

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

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

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

Внимание

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

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

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

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

4. Выберите + Добавить в верхнем меню, а затем выберите Добавить назначение роли в появившемся раскрывающемся меню.

   

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

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

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

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

Строка подключения

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

При создании нового пространства имен автоматически создается начальная политика подписанного URL-адреса (SAS), первичный и вторичный ключи, а также первичная и вторичная строки подключения, каждая из которых предоставляет полный контроль над всеми аспектами пространства имен. Сведения о том, как в дальнейшем создавать правила с ограниченными правами для постоянных отправителей и получателей, см. в статье Аутентификация и авторизация в служебной шине.

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

1. На странице Пространство имен служебной шины выберите Политики общего доступа в меню слева.

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

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

   

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

Установка пакетов с помощью pip

Без пароля (рекомендуется)

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

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

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

Строка соединения

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

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

   pip install azure-servicebus
   

Отправка сообщений в очередь

В следующем примере кода показано, как отправить сообщение в очередь. Откройте любимый редактор, например Visual Studio Code, создайте файл send.py и добавьте в него следующий код.

Без пароля (рекомендуется)

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

Получение сообщений из очереди

В следующем примере кода показано, как получать сообщения из очереди. Отображаемый код получает новые сообщения до тех пор, пока он не получит новые сообщения в течение 5 (max_wait_time) секунд.

Откройте любимый редактор, например Visual Studio Code, создайте файл recv.py и добавьте в него следующий код.

Без пароля (рекомендуется)

1. Как и в примере отправки, добавьте 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. Как и в примере отправки, добавьте 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())
   

Выполнить приложение

Откройте командную строку с 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. Если количество не отображается, обновите страницу, подождав несколько минут.

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

Следующие шаги

Ознакомьтесь со следующими примерами и документацией:

• Клиентская библиотека Служебной шины Azure для Python
• Примеры.
  • В папке sync_samples приведены примеры, демонстрирующие взаимодействие со служебной шиной в синхронном режиме.
  • В папке async_samples приведены примеры, демонстрирующие взаимодействие со служебной шиной в асинхронном режиме. Проходя краткое руководство, вы использовали этот метод.
• Справочная документация по использованию пакета azure-servicebus