適用于 Python 的 Azure 服務匯流排 用戶端程式庫 - 7.11.4 版
Azure 服務匯流排是高效能的雲端管理傳訊服務，可提供分散式傳送者和接收者之間的即時和容錯通訊。
服務匯流排提供多個機制來進行非同步高度可靠的通訊，例如結構化先進先出傳訊、發佈/訂閱功能，以及在需求成長時輕鬆調整的能力。
使用適用于 Python 的服務匯流排用戶端程式庫，在應用程式和服務之間通訊，並實作非同步傳訊模式。
- 建立服務匯流排命名空間、佇列、主題和訂用帳戶，並修改其設定。
- 在您的服務匯流排通道內傳送和接收訊息。
- 利用訊息鎖定、會話和寄不出的信件功能來實作複雜的傳訊模式。
| 原始程式碼Package (PyPi) | 封裝 (Conda) | API 參考檔 | 產品檔 | 樣品 | Changelog
注意：如果您使用 0.50 版或更低版本，而且想要移轉至此套件的最新版本，請參閱我們的 移轉指南，以從服務匯流排 V0.50 移至服務匯流排 V7。
開始使用
安裝套件
使用pip安裝適用于 Python 的 Azure 服務匯流排 用戶端程式庫：
pip install azure-servicebus
必要條件：
若要使用此套件，您必須具有：
如果您需要 Azure 服務匯流排命名空間，您可以透過 Azure 入口網站加以建立。 如果您不想使用圖形化入口網站 UI，您可以使用 Azure CLI，透過Cloud Shell或 Azure CLI 在本機執行，以使用此 Azure CLI 命令來建立：
az servicebus namespace create --resource-group <resource-group-name> --name <servicebus-namespace-name> --location <servicebus-namespace-location>
驗證用戶端
與服務匯流排的互動會從 類別的 ServiceBusClient 實例開始。 您需要具有 SAS 金鑰的連接字串，或命名空間和其中一個帳戶金鑰來具現化用戶端物件。
如需如何透過任一方法進行驗證的示範，請尋找以下連結的範例。
從 連接字串 建立用戶端
- 若要取得所需的認證，您可以 (遵循取得服務匯流排連接字串連接字串) 的逐步指南，使用連結範例頂端的 Bash Shell) 來填入環境變數，連接字串 (您也可以依照取得服務匯流排連接字串) 的逐步指南，在Azure入口網站中找到這些值。
使用 azure-identity 程式庫建立用戶端：
- 此建構函式會採用服務匯流排實例的完整命名空間，以及實作 TokenCredential 通訊協定的認證。
TokenCredentialazure-identity 套件中有可用的通訊協定實作。 完整命名空間的格式為<yournamespace.servicebus.windows.net>。 - 若要使用 所提供的
azure-identity認證類型，請安裝套件：pip install azure-identity - 此外，若要使用非同步 API，您必須先安裝非同步傳輸，例如
aiohttp：pip install aiohttp - 使用 Azure Active Directory 時，您的主體必須獲指派允許存取服務匯流排的角色，例如Azure 服務匯流排資料擁有者角色。 如需搭配服務匯流排使用 Azure Active Directory 授權的詳細資訊，請參閱 相關聯的檔。
注意： 用戶端可以在沒有內容管理員的情況下初始化，但必須透過 client.close () 手動關閉，才能不流失資源。
重要概念
初始化 ServiceBusClient 之後，您就可以與服務匯流排命名空間內的主要資源類型互動，其中多個可以存在，以及發生實際訊息傳輸時，命名空間通常會作為應用程式容器：
佇列：允許傳送和接收訊息。 通常用於點對點通訊。
主題：與佇列相反，主題更適合發佈/訂閱案例。 主題可以傳送至 ，但需要一個訂用帳戶，其中可以平行多個，以取用來源。
訂用帳戶：從主題取用的機制。 每個訂用帳戶都是獨立的，並接收傳送至主題的每個訊息複本。 規則和篩選可用來量身打造特定訂用帳戶所接收的訊息。
如需這些資源的詳細資訊，請參閱什麼是Azure 服務匯流排？。
若要與這些資源互動，您應該熟悉下列 SDK 概念：
ServiceBusClient：這是使用者第一次初始化以連線至服務匯流排命名空間的物件。 若要與佇列、主題或訂用帳戶互動，一個會從此用戶端繁衍傳送者或接收者。
ServiceBusSender：若要將訊息傳送至佇列或主題，則會使用 實例的
ServiceBusClient對應get_queue_sender或get_topic_sender方法，如這裡所示。ServiceBusReceiver：若要從佇列或訂用帳戶接收訊息，其中一個會使用實例的
ServiceBusClient對應get_queue_receiver或get_subscription_receiver方法，如下所示。ServiceBusMessage：傳送時，這是您將建構以包含承載的類型。 接收時，這是您將存取承載的位置。
執行緒安全
我們並不保證 ServiceBusClient、ServiceBusSender 和 ServiceBusReceiver 是安全線程。 不建議跨執行緒重複使用這些實例。 執行中的應用程式會以安全線程的方式使用這些類別。
範例
下列各節提供數個程式碼片段，涵蓋一些最常見的服務匯流排工作，包括：
若要執行管理工作，例如建立和刪除佇列/主題/訂用帳戶，請使用 azure-mgmt-servicebus 程式庫， 這裡提供。
請參閱範例目錄中的進一步 範例 ，示範常見的服務匯流排案例，例如傳送、接收、會話管理和訊息處理。
傳送訊息至佇列
注意： 請參閱 這裡的參考檔。
本範例會將單一訊息和訊息陣列傳送至假設已經存在的佇列，並透過 Azure 入口網站 或 az 命令建立。
from azure.servicebus import ServiceBusClient, ServiceBusMessage
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_sender(queue_name) as sender:
# Sending a single message
single_message = ServiceBusMessage("Single message")
sender.send_messages(single_message)
# Sending a list of messages
messages = [ServiceBusMessage("First message"), ServiceBusMessage("Second message")]
sender.send_messages(messages)
注意： 您可以使用 方法來排程訊息以進行延遲傳遞
ServiceBusSender.schedule_messages()，或藉ServiceBusMessage.scheduled_enqueue_time_utc由在呼叫之前指定ServiceBusSender.send_messages()
如需排程和排程取消的詳細資訊，請參閱 這裡的範例。
從佇列接收訊息
若要從佇列接收，您可以透過 receiver.receive_messages() 執行臨機操作接收，或持續透過接收者本身接收。
透過逐一查看 ServiceBusReceiver 從佇列接收訊息
from azure.servicebus import ServiceBusClient
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
# max_wait_time specifies how long the receiver should wait with no incoming messages before stopping receipt.
# Default is None; to receive forever.
with client.get_queue_receiver(queue_name, max_wait_time=30) as receiver:
for msg in receiver: # ServiceBusReceiver instance is a generator.
print(str(msg))
# If it is desired to halt receiving early, one can break out of the loop here safely.
注意： (收到
receive_mode=PEEK_LOCK的任何訊息都是預設值，而替代RECEIVE_AND_DELETE在收到時立即從佇列中移除訊息，) 具有必須透過receiver.renew_message_lock更新的鎖定，如果處理時間超過鎖定持續時間，則必須透過 更新。 請參閱 AutoLockRenewer ，以取得協助程式自動在背景中執行這項作業。 鎖定持續時間是在 Azure 中設定于佇列或主題本身。
透過 ServiceBusReceiver.receive_messages () 從佇列接收訊息
注意：
ServiceBusReceiver.receive_messages()透過臨機操作方法呼叫接收單一或限制的訊息清單，而不是從產生器永久接收。 它一律會傳回清單。
from azure.servicebus import ServiceBusClient
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_receiver(queue_name) as receiver:
received_message_array = receiver.receive_messages(max_wait_time=10) # try to receive a single message within 10 seconds
if received_message_array:
print(str(received_message_array[0]))
with client.get_queue_receiver(queue_name) as receiver:
received_message_array = receiver.receive_messages(max_message_count=5, max_wait_time=10) # try to receive maximum 5 messages in a batch within 10 seconds
for message in received_message_array:
print(str(message))
在此範例中，max_message_count會宣告在按秒指定的max_wait_time之前嘗試接收的訊息數目上限。
注意： 也應該注意，
ServiceBusReceiver.peek_messages()與接收不一樣，因為它不會鎖定正在查看的訊息，因此無法加以解決。
從已啟用會話的佇列傳送和接收訊息
會話會在佇列或訂用帳戶頂端提供先進先出和單一接收者語意。 雖然實際的接收語法相同，但初始化稍有不同。
from azure.servicebus import ServiceBusClient, ServiceBusMessage
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_SESSION_QUEUE_NAME']
session_id = os.environ['SERVICE_BUS_SESSION_ID']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_sender(queue_name) as sender:
sender.send_messages(ServiceBusMessage("Session Enabled Message", session_id=session_id))
# If session_id is null here, will receive from the first available session.
with client.get_queue_receiver(queue_name, session_id=session_id) as receiver:
for msg in receiver:
print(str(msg))
注意：從會話接收的訊息不需要其鎖定更新，就像非會話接收者一樣;相反地，鎖定管理會在工作階段層級發生，且會話鎖定可能會更新
receiver.session.renew_lock()
使用 主題 和訂用 帳戶
主題和訂用帳戶提供用於傳送和接收訊息的佇列替代方案。 如需更整體的詳細資料，請參閱 這裡的 檔，以及這些與佇列有何不同。
from azure.servicebus import ServiceBusClient, ServiceBusMessage
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
topic_name = os.environ['SERVICE_BUS_TOPIC_NAME']
subscription_name = os.environ['SERVICE_BUS_SUBSCRIPTION_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_topic_sender(topic_name) as sender:
sender.send_messages(ServiceBusMessage("Data"))
# If session_id is null here, will receive from the first available session.
with client.get_subscription_receiver(topic_name, subscription_name) as receiver:
for msg in receiver:
print(str(msg))
在收據之後結算訊息
從佇列接收時，您可以對收到的訊息採取多個動作。
注意：您只能將以模式接收
ServiceBusReceiveMode.PEEK_LOCK的物件結算ServiceBusReceivedMessage， (這是預設) 。ServiceBusReceiveMode.RECEIVE_AND_DELETEmode 會從收到時從佇列中移除訊息。ServiceBusReceivedMessage從peek_messages()傳回的訊息無法解決，因為訊息鎖定不會像上述接收方法一樣。
如果訊息有如上所述的鎖定，如果訊息鎖定已過期，settlement 將會失敗。
如果處理時間超過鎖定持續時間，則必須在到期之前透過 receiver.renew_message_lock 維護。
鎖定持續時間是在 Azure 中設定于佇列或主題本身。
請參閱 AutoLockRenewer ，以取得協助程式自動在背景中執行這項作業。
完成
宣告要成功完成的訊息處理，從佇列中移除訊息。
from azure.servicebus import ServiceBusClient
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_receiver(queue_name) as receiver:
for msg in receiver:
print(str(msg))
receiver.complete_message(msg)
放棄
放棄處理訊息一段時間，立即將訊息傳回佇列，以供另一個 (或相同的) 接收者挑選。
from azure.servicebus import ServiceBusClient
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_receiver(queue_name) as receiver:
for msg in receiver:
print(str(msg))
receiver.abandon_message(msg)
DeadLetter
將訊息從主要佇列傳送至特殊的「寄不出的信件子佇列」，以便使用 ServiceBusClient.get_<queue|subscription>_receiver 函式搭配 參數 sub_queue=ServiceBusSubQueue.DEAD_LETTER 存取，並像任何其他接收者一樣取用。 (請參閱 這裡的 範例)
from azure.servicebus import ServiceBusClient
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_receiver(queue_name) as receiver:
for msg in receiver:
print(str(msg))
receiver.dead_letter_message(msg)
推遲
延遲與先前的結算方法不同。 它會藉由將訊息設定為必須在呼叫 ServiceBusReceiver.receive_deferred_messages 中以序號接收，以防止訊息直接從佇列接收， (請參閱這裡的 範例)
from azure.servicebus import ServiceBusClient
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_receiver(queue_name) as receiver:
for msg in receiver:
print(str(msg))
receiver.defer_message(msg)
自動更新訊息或會話鎖定
注意： 請參閱 自動鎖定更新的參考檔。
AutoLockRenewer是一種簡單的方法，可讓您的訊息或會話在長時間內保持鎖定狀態，如果呼叫 receiver.renew_message_lock/receiver.session.renew_lock 不切實際或不想要的話。
就內部而言，如果物件即將到期，建立並行監看狗進行鎖定更新的速記並不多。
它應該使用如下：
- 訊息鎖定自動更新
from azure.servicebus import ServiceBusClient, AutoLockRenewer
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
# Can also be called via "with AutoLockRenewer() as renewer" to automate closing.
renewer = AutoLockRenewer()
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_receiver(queue_name) as receiver:
for msg in receiver.receive_messages():
renewer.register(receiver, msg, max_lock_renewal_duration=60)
# Do your application logic here
receiver.complete_message(msg)
renewer.close()
- 會話鎖定自動更新
from azure.servicebus import ServiceBusClient, AutoLockRenewer
import os
connstr = os.environ['SERVICE_BUS_CONNECTION_STR']
session_queue_name = os.environ['SERVICE_BUS_SESSION_QUEUE_NAME']
session_id = os.environ['SERVICE_BUS_SESSION_ID']
# Can also be called via "with AutoLockRenewer() as renewer" to automate closing.
renewer = AutoLockRenewer()
with ServiceBusClient.from_connection_string(connstr) as client:
with client.get_queue_receiver(session_queue_name, session_id=session_id) as receiver:
renewer.register(receiver, receiver.session, max_lock_renewal_duration=300) # Duration for how long to maintain the lock for, in seconds.
for msg in receiver.receive_messages():
# Do your application logic here
receiver.complete_message(msg)
renewer.close()
如果基於任何原因自動續約中斷或失敗，可以透過 auto_renew_error 更新物件上的 屬性來觀察，或透過在更新程式初始化時傳遞回呼至 on_lock_renew_failure 參數。
嘗試採取動作時也會顯示資訊清單 (，例如在指定的物件上完成訊息) 。
疑難排解
記錄
- 啟用
azure.servicebus記錄器以從程式庫收集追蹤。 - 啟用
uamqp記錄器以從基礎 uAMQP 程式庫收集追蹤。 - 在建立用戶端時設定
logging_enable=True，以啟用 AMQP 畫面層級追蹤。 - 在某些情況下，您可能會將記錄視為
uamqp太詳細資訊。 若要隱藏不必要的記錄，請將下列程式碼片段新增至程式碼頂端：
import logging
# The logging levels below may need to be changed based on the logging that you want to suppress.
uamqp_logger = logging.getLogger('uamqp')
uamqp_logger.setLevel(logging.ERROR)
# or even further fine-grained control, suppressing the warnings in uamqp.connection module
uamqp_connection_logger = logging.getLogger('uamqp.connection')
uamqp_connection_logger.setLevel(logging.ERROR)
逾時
使用者應該在程式庫中注意各種逾時。
- 10 分鐘的服務端連結關閉：一旦開啟連結，就會在閒置 10 分鐘後關閉，以保護服務免于資源外泄。 這應該對使用者而言大致上是透明的，但如果您注意到在這類持續時間之後發生重新連線，這就是原因。 在連結上執行任何作業，包括管理作業，將會延長此逾時。
- max_wait_time：在建立接收者時，或在呼叫
receive_messages()時提供，接收訊息之後不會有任何流量後停止的時間。 這同時適用于命令receive_messages()式以及產生器樣式接收在結束之前，如果沒有任何訊息，則會執行的長度。 傳遞 None (預設) 會永遠等候，直到沒有採取其他動作時，直到 10 分鐘閾值為止。
注意：如果訊息或會話的處理時間夠長，就足以造成逾時，作為手動呼叫
receiver.renew_message_lock/receiver.session.renew_lock的替代方法，則可以利用AutoLockRenewer上述詳細。
常見的例外狀況
服務匯流排 API 會在 azure.servicebus.exceptions 中產生下列例外狀況：
- ServiceBusConnectionError： 與服務的連線發生錯誤。 這可能是因為暫時性網路問題或服務問題所造成。 建議您重試。
- ServiceBusAuthorizationError： 授權服務連線時發生錯誤。 這可能是因為認證沒有執行作業的正確許可權所造成。 建議您檢查認證的許可權。
- ServiceBusAuthenticationError： 驗證服務的連線時發生錯誤。 這可能是因為認證不正確所造成。 建議您檢查認證。
- OperationTimeoutError： 這表示服務未在預期的時間內回應作業。 這可能是因為暫時性網路問題或服務問題所造成。 服務可能或可能尚未順利完成要求;狀態未知。 建議您嘗試確認目前的狀態，並視需要重試。
- MessageSizeExceededError： 這表示訊息內容大於服務匯流排框架大小。
當批次中傳送太多服務匯流排訊息，或傳遞至 主體
Message的內容太大時，就會發生這種情況。 建議您減少在批次中傳送的訊息計數，或傳遞至單ServiceBusMessage一 的內容大小。 - MessageAlreadySettled： 這表示無法結算訊息。 嘗試解決已解決的訊息時，可能會發生這種情況。
- MessageLockLostError： 訊息上的鎖定已過期，且已釋回佇列。
必須再次收到，才能加以解決。
您應該注意訊息的鎖定持續時間，並在處理時間很長時，在到期前持續更新鎖定。
AutoLockRenewer有助於讓訊息保持自動更新的鎖定。 - SessionLockLostError： 會話上的鎖定已過期。
無法再解決所有已接收的未排定訊息。
如有需要再次接收訊息，建議重新連線到會話。
您應該注意會話的鎖定持續時間，並在處理時間很長時，在到期前持續更新鎖定。
AutoLockRenewer有助於保持會話的鎖定自動更新。 - MessageNotFoundError： 嘗試接收具有特定序號的訊息。 找不到此訊息。 請確定尚未收到訊息。 檢查寄不出信件佇列，查看訊息是否已停止傳送。
- MessagingEntityNotFoundError： 與作業相關聯的實體不存在或已刪除。 請確定實體存在。
- MessagingEntityDisabledError： 在停用的實體上要求執行時間作業。 請啟動實體。
- ServiceBusQuotaExceededError： 傳訊實體已達到其允許大小上限，或已超過命名空間的連線數目上限。 從實體或其子佇列接收訊息，在實體中建立空間。
- ServiceBusServerBusyError： 服務目前無法處理要求。 用戶端可以等待一段時間，然後再重試作業。
- ServiceBusCommunicationError： 用戶端無法建立服務匯流排的連線。 確定提供的主機名稱正確，且主機可以連線。 如果您的程式碼在具有防火牆/Proxy 的環境中執行，請確定服務匯流排網域/IP 位址和埠的流量不會遭到封鎖。
- SessionCannotBeLockedError： 嘗試連線到具有特定會話識別碼的會話，但會話目前由另一個用戶端鎖定。 確定其他用戶端已解除鎖定工作階段。
- AutoLockRenewFailed： 嘗試在背景中更新訊息或會話的鎖定失敗。
當 使用的
AutoLockRenewer接收者已關閉或更新的鎖定已過期時，就會發生這種情況。 建議您藉由接收訊息或再次連線到會話實體來重新註冊可更新的訊息或會話。 - AutoLockRenewTimeout： 配置來更新訊息或會話鎖定的時間已過。 您可以重新註冊想要自動更新鎖定的物件，或事先延長逾時。
- ServiceBusError： 所有其他服務匯流排相關錯誤。 這是上述所有錯誤的根錯誤類別。
如需常見例外狀況類型的詳細描述，請檢視 例外狀況參考檔 。
下一步
更多的程式碼範例
請參閱範例目錄中的進一步 範例 ，示範常見的服務匯流排案例，例如傳送、接收、會話管理和訊息處理。
其他文件
如需有關服務匯流排服務的詳細資訊，請參閱有關 docs.microsoft.com 的服務匯流排檔 。
管理功能和檔
如需針對 ServiceBus 執行管理作業的使用者 (建立佇列/主題/等、改變篩選規則、列舉實體) 請參閱 API 檔的 azure-mgmt-servicebus 檔 。 您也可以 在這裡 找到 Terse 使用範例。
純 Python AMQP 傳輸和回溯相容性支援
Azure 服務匯流排用戶端程式庫現在是以純 Python AMQP 實作為基礎。 uAMQP 已視需要移除相依性。
若要作為基礎傳輸使用 uAMQP ：
- 使用 pip 安裝
uamqp。
$ pip install uamqp
- 在用戶端建構期間傳遞
uamqp_transport=True。
from azure.servicebus import ServiceBusClient
connection_str = '<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>'
queue_name = '<< NAME OF THE QUEUE >>'
client = ServiceBusClient.from_connection_string(
connection_str, uamqp_transport=True
)
注意： message 先前公開 的 uamqp.Message 屬性 ServiceBusReceivedMessageServiceBusMessage/ServiceBusMessageBatch/ 已被取代。
已導入屬性所 message 傳回的「舊版」物件，以協助轉換。
從來源建置 uAMQP 滾輪
azure-servicebus 取決於 AMQP 通訊協定實作的 uAMQP 。
uAMQP 方向盤會提供給大部分的主要作業系統，並在安裝 時自動安裝 azure-servicebus 。
如果 uAMQP 是要用來做為 的基礎 AMQP 通訊協定實作 azure-servicebus ，則可以針對大部分的主要作業系統找到 uAMQP 方向盤。
如果您在未提供 uAMQP 方向盤的平臺上執行，請遵循 如果您想要使用 uAMQP ，而且您是在未提供 uAMQP 方向盤的平臺上執行，請遵循 uAMQP 安裝 指引以從來源安裝。
參與
此專案歡迎參與和提供建議。 大部分的參與都要求您同意「參與者授權合約 (CLA)」，宣告您有權且確實授與我們使用投稿的權利。 如需詳細資料，請前往 https://cla.microsoft.com 。
當您提交提取要求時，CLA Bot 會自動判斷您是否需要提供 CLA，並適當地裝飾 PR (例如標籤、註解)。 請遵循 bot 提供的指示。 您只需要使用我們的 CLA 在所有存放庫上執行此動作一次。
此專案採用 Microsoft Open Source Code of Conduct (Microsoft 開放原始碼管理辦法)。 如需詳細資訊，請參閱管理辦法常見問題集，如有任何其他問題或意見請連絡 opencode@microsoft.com。
意見反應
即將登場：在 2024 年，我們將逐步淘汰 GitHub 問題作為內容的意見反應機制，並將它取代為新的意見反應系統。 如需詳細資訊，請參閱：https://aka.ms/ContentUserFeedback。
提交並檢視相關的意見反應