將訊息傳送至 Azure 服務匯流排 主題，並從訂用帳戶接收主題的訊息 （Python）

將角色指派給 Microsoft Entra 使用者

在本機開發時，請確定連線到 Azure 服務匯流排 的用戶帳戶具有正確的許可權。 您需要 Azure 服務匯流排 數據擁有者角色，才能傳送和接收訊息。 若要指派此角色，您需要使用者存取 管理員 istrator 角色，或包含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] 視窗中，選取 [主要 連線 ion String] 旁的 [複製] 按鈕，將 連接字串 複製到剪貼簿以供稍後使用。 將此值貼到 記事本 或其他暫存位置。

   

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

若要遵循本快速入門，請使用無密碼驗證和您自己的 Azure 帳戶：

• 安裝 Azure CLI。
• 在終端機或命令提示字元 az login中使用 您的 Azure 帳戶登入。
• 當您稍後在教學課程中將適當的角色新增至資源時，請使用相同的帳戶。
• 在相同的終端機或命令提示字元中執行教學課程程序代碼。

若要使用無密碼程式代碼，您必須指定：

• 完整服務總線命名空間，例如：<service-bus-namespace.servicebus.windows.net>
• 主題名稱
• 訂閱名稱

若要遵循本快速入門，請使用 連接字串 進行驗證，您不會使用自己的 Azure 帳戶。 相反地，您將針對服務總線命名空間使用 連接字串。

若要使用連線程式代碼，您必須指定：

• 連接字串
• 主題名稱
• 訂閱名稱

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

   重要

   • 將取代FULLY_QUALIFIED_NAMESPACE為您 服務匯流排 命名空間的完整命名空間。
   • 將取代 TOPIC_NAME 為主題的名稱。

   在上述程式代碼中，您已使用 Azure 身分識別用戶端連結庫的 DefaultAzureCredential 類別。 當應用程式在開發期間於本機執行時， DefaultAzureCredential 將會使用您登入 Azure CLI 的帳戶自動探索並驗證至 Azure。 將應用程式部署至 Azure 時， DefaultAzureCredential 可以透過受控識別向 Microsoft Entra 識別碼驗證您的應用程式，而不需要變更任何程式代碼。

3. 新增方法以傳送單一訊息。

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

   傳送者是物件，做為您建立之主題的用戶端。 您稍後會建立它，並將 做為自變數傳送至此函式。

4. 新增方法以傳送訊息清單。

   async def send_a_list_of_messages(sender):
       # Create a list of messages
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       # send the list of messages to the topic
       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 topic
           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 Topic Sender object to send messages to the topic
           sender = servicebus_client.get_topic_sender(topic_name=TOPIC_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()
   
   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_STRING"
   TOPIC_NAME = "TOPIC_NAME"
   

   重要

   • 將取代NAMESPACE_CONNECTION_STRING為您的命名空間 連接字串。
   • 將取代 TOPIC_NAME 為主題的名稱。

3. 新增方法以傳送單一訊息。

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

   傳送者是物件，做為您建立之主題的用戶端。 您稍後會建立它，並將 做為自變數傳送至此函式。

4. 新增方法以傳送訊息清單。

   async def send_a_list_of_messages(sender):
       # Create a list of messages
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       # send the list of messages to the topic
       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 topic
           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 Topic Sender object to send messages to the topic
           sender = servicebus_client.get_topic_sender(topic_name=TOPIC_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)
   
   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. 類似於傳送範例 add import 語句、定義您應該以您自己的值取代的常數，以及定義認證。

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

2. 建立 服務匯流排 客戶端，然後建立要接收訊息的訂用帳戶接收者物件。

   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:
   
           async with servicebus_client:
               # get the Subscription Receiver object for the subscription
               receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_NAME, 
               subscription_name=SUBSCRIPTION_NAME, max_wait_time=5)
               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 subscription
                       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_STRING"
   SUBSCRIPTION_NAME = "SUBSCRIPTION_NAME"
   TOPIC_NAME = "TOPIC_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 Subscription Receiver object for the subscription
               receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_NAME, 
               subscription_name=SUBSCRIPTION_NAME, max_wait_time=5)
               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 subscription
                       await receiver.complete_message(msg)
   

3. 呼叫 run 方法。

   asyncio.run(run())