在 Azure 服務匯流排佇列 (Python) 中傳送和接收訊息

若要搭配您自己的 Azure 帳戶使用此快速入門：

• 安裝 Azure CLI，為開發人員電腦提供無密碼驗證。
• 在終端機或命令提示字元中，使用 az login 登入您的 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 服務匯流排資料擁有者角色。 其可讓在使用者帳戶內容中執行的應用程式將訊息傳送至佇列或主題，以及從佇列或主題的訂用帳戶接收訊息。

1. 如果沒有在 Azure 入口網站中開啟 [服務匯流排命名空間] 頁面，請使用主要搜尋列或左側導覽找出您的服務匯流排命名空間。

2. 在概觀頁面上，從左側功能表中選取 [存取控制 (IAM)]。

3. 在 [存取控制 (IAM)] 頁面上，選取 [角色指派] 索引標籤。

4. 從頂端功能表選取 [+ 新增]，然後從產生的下拉功能表中選取 [新增角色指派]。

   

5. 使用搜尋方塊，從結果篩選出所需的角色。 在此範例中，搜尋 Azure Service Bus Data Owner 並選取相符的結果。 接著，選擇 [下一步]。

6. 在 [存取權指派對象為] 下，選取 [使用者、群組或服務主體]，然後選擇 [+ 選取成員]。

7. 在對話方塊中，搜尋 Microsoft Entra 使用者名稱 (通常是您的 user@domain 電子郵件地址)，然後在對話方塊底部選擇 [選取]。

8. 選取 [檢閱 + 指派] 以移至最終頁面，然後再次選取 [檢閱 + 指派] 以完成此程序。

取得連接字串

建立新命名空間時會自動產生初始共用存取簽章 (SAS) 原則與主要和次要金鑰，以及主要和次要連接字串，其會授與命名空間所有層面的完整控制權。 請參閱服務匯流排驗證與授權，以了解如何建立進一步的規則，對一般傳送者和接收者的權限施加更多限制。

用戶端可以使用連接字串來連線到服務匯流排命名空間。 若要複製命名空間的主要連接字串，請遵循下列步驟：

1. 在 [服務匯流排命名空間] 頁面上，選取左側功能表的 [共用存取原則]。

2. 在 [共用存取原則] 頁面上，選取 [RootManageSharedAccessKey]。

3. 在 [原則：RootManageSharedAccessKey] 視窗中，選取在 [主要連接字串] 旁邊的複製按鈕，將連接字串複製到您的剪貼簿以供後續使用。 將此值貼到記事本或一些其他暫存位置。

   

   您可使用此頁面來複製主要金鑰、次要金鑰、主要連接字串和次要連接字串。

1. 若要安裝本服務匯流排教學課程所需的 Python 套件，請開啟在路徑中有 Python 的命令提示字元，並將目錄切換至您要用來存放範例的資料夾。

2. 安裝下列套件：

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

1. 若要安裝本服務匯流排教學課程所需的 Python 套件，請開啟在路徑中有 Python 的命令提示字元，並將目錄切換至您要用來存放範例的資料夾。

2. 安裝下列套件：

   pip install azure-servicebus
   

1. 新增 import 陳述式。

   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 陳述式。

   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. 類似於傳送範例，請新增 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())