EventHubConsumerClient 類別

參考

EventHubConsumerClient 類別會定義從Azure 事件中樞服務接收事件的高階介面。

EventHubConsumerClient的主要目標是使用負載平衡和檢查點，從 EventHub 的所有分割區接收事件。

當多個 EventHubConsumerClient 實例針對相同的事件中樞、取用者群組和檢查點位置執行時，分割區會平均分散到其中。

若要啟用負載平衡和保存的檢查點，必須在建立 EventHubConsumerClient時設定checkpoint_store。如果未提供檢查點存放區，則會在記憶體內部維護檢查點。

當您呼叫 EventHubConsumerClient 方法 receive () 或 receive_batch () 時，也可以從特定分割區接收，並指定partition_id。負載平衡無法在單一資料分割模式中運作。但是，如果已設定checkpoint_store，使用者仍然可以儲存檢查點。

繼承: azure.eventhub.aio._client_base_async.ClientBaseAsync

EventHubConsumerClient

建構函式

EventHubConsumerClient(fully_qualified_namespace: str, eventhub_name: str, consumer_group: str, credential: CredentialTypes, **kwargs: Any)

參數

fully_qualified_namespace: str

必要

事件中樞命名空間的完整主機名稱。命名空間格式為： .servicebus.windows.net。

eventhub_name: str

必要

要連接用戶端的特定事件中樞路徑。

consumer_group: str

必要

從這個取用者群組的事件中樞接收事件。

credential: AsyncTokenCredential 或 AzureSasCredential 或 AzureNamedKeyCredential

必要

用於驗證的認證物件，會實作用於取得權杖的特定介面。它會接受 EventHubSharedKeyCredential 由 azure-identity 程式庫所產生的認證物件，以及實作 *get_token (自我範圍) 方法的物件。

logging_enable: bool

是否要將網路追蹤記錄輸出至記錄器。預設值為 False。

auth_timeout: float

等候服務授權權杖的秒數。預設值為 60 秒。如果設定為 0，則不會從用戶端強制執行逾時。

user_agent: str

如果指定，這會新增到使用者代理程式字串前面。

retry_total: int

發生錯誤時，嘗試取消復原失敗作業的嘗試總數。預設值為 3。接收中 retry_total 的內容特別：接收方法是由在每次反覆運算中呼叫內部接收方法的 while 迴圈實作。在接收案例中， retry_total 指定 while-loop 中內部接收方法引發錯誤之後重試的數目。如果重試嘗試用盡，如果提供) 錯誤資訊，則會呼叫 ( on_error回呼。如果提供) 而 (建立新的內部分割區取用者，則會on_partition_close (在提供) 時呼叫失敗的內部分割區取用者，on_partition_initialize) 繼續接收時呼叫失敗的內部分割區取用者。

retry_backoff_factor: float

第二次嘗試之後要套用的輪詢因素， (大部分的錯誤都會立即由第二次嘗試解決，而不會延遲) 。在固定模式中，重試原則一律會睡眠 {backoff factor}。在「指數」模式中，重試原則會睡眠： {backoff factor} * (2 ** ({total retries} - 1) ) 秒。如果backoff_factor為 0.1，則重試會在重試之間睡眠 [0.0s， 0.2s， 0.4s， ...]。預設值為 0.8。

retry_backoff_max: float

退班時間上限。預設值為 120 秒， (2 分鐘) 。

retry_mode: str

重試嘗試之間的延遲行為。支援的值為 'fixed' 或 'exponential'，其中預設值為 'exponential'。

idle_timeout: float

逾時，以秒為單位，在此之後，如果沒有進一步的活動，此用戶端將會關閉基礎連線。根據預設，此值為 None，表示除非服務起始，否則用戶端不會因為閒置而關閉。

transport_type: TransportType

將用於與事件中樞服務通訊的傳輸通訊協定類型。預設值為 TransportType.Amqp ，在此情況下會使用埠 5671。如果網路環境中無法使用/封鎖埠 5671，則可改用 TransportType.AmqpOverWebsocket 來使用埠 443 進行通訊。

http_proxy

HTTP Proxy 設定。這必須是具有下列索引鍵的字典： 'proxy_hostname' ( str 值) 和 'proxy_port' (int 值) 。

checkpoint_store: Optional[CheckpointStore]

在接收事件時儲存分割區負載平衡和檢查點資料的管理員。檢查點存放區將會用於從所有分割區或單一分割區接收的兩種情況。在後者的情況下，不會套用負載平衡。如果未提供檢查點存放區，則會在記憶體內部維護檢查點，而且 EventHubConsumerClient 實例將會收到事件，而不會進行負載平衡。

load_balancing_interval: float

負載平衡啟動時。這是兩個負載平衡評估之間的間隔，以秒為單位。預設值為 30 秒。

partition_ownership_expiration_interval: float

分割區擁有權會在此秒數之後過期。每個負載平衡評估都會自動延長擁有權到期時間。預設值為 6 * load_balancing_interval，也就是使用預設load_balancing_interval為 30 秒時的 180 秒。

load_balancing_strategy: str 或 LoadBalancingStrategy

當負載平衡開始時，它會使用此策略來宣告和平衡分割區擁有權。使用「gredy」或 LoadBalancingStrategy.GREDY 作為窮盡策略，針對每個負載平衡評估，將會擷取平衡負載所需的未重載分割區數目。針對平衡策略使用「balanced」或 LoadBalancingStrategy.BALANCED ，針對每個負載平衡評估，只會宣告其他 EventHubConsumerClient未宣告的一個分割區。如果 EventHub 的所有分割區都由其他 EventHubConsumerClient 宣告，而且此用戶端宣告了太少的資料分割，則不論負載平衡策略為何，此用戶端都會從其他用戶端竊取一個分割區，預設會使用窮盡策略。

custom_endpoint_address: Optional[str]

用來建立事件中樞服務的連線的自訂端點位址，允許透過任何應用程式閘道或其他主機環境所需的路徑路由傳送網路要求。預設值為 None。格式會像「sb:// < custom_endpoint_hostname > ： < custom_endpoint_port > 」。如果未在 custom_endpoint_address中指定埠，預設會使用埠 443。

connection_verify: Optional[str]

SSL 憑證之自訂CA_BUNDLE檔案的路徑，用來驗證連線端點的身分識別。預設值為 None，在此情況下會使用 certifi.where () 。

uamqp_transport: bool

是否要使用 uamqp 程式庫作為基礎傳輸。預設值為 False，而純 Python AMQP 程式庫將作為基礎傳輸。

socket_timeout: float

連接上基礎通訊端在傳送和接收資料之前應該等候的時間，以秒為單位。TransportType.Amqp 的預設值為 0.2，而 TransportType.AmqpOverWebsocket 的預設值為 1。如果 EventHubsConnectionError 錯誤是因為寫入逾時而發生，可能需要傳入大於預設值。這適用于進階使用案例，而且預設值通常就已足夠。

範例

建立 EventHubConsumerClient 的新實例。


   import os
   from azure.eventhub.aio import EventHubConsumerClient, EventHubSharedKeyCredential

   fully_qualified_namespace = os.environ['EVENT_HUB_HOSTNAME']
   eventhub_name = os.environ['EVENT_HUB_NAME']
   shared_access_policy = os.environ['EVENT_HUB_SAS_POLICY']
   shared_access_key = os.environ['EVENT_HUB_SAS_KEY']

   consumer = EventHubConsumerClient(fully_qualified_namespace=fully_qualified_namespace,
                                     consumer_group='$Default',
                                     eventhub_name=eventhub_name,
                                     credential=EventHubSharedKeyCredential(shared_access_policy, shared_access_key))

方法

close	停止從事件中樞擷取事件，並關閉基礎 AMQP 連線和連結。
from_connection_string	從連接字串建立 EventHubConsumerClient。
get_eventhub_properties	取得事件中樞的屬性。傳回字典中的索引鍵包括： eventhub_name str) ( created_at (UTC datetime.datetime) partition_ids (list[str])
get_partition_ids	取得事件中樞的資料分割識別碼。
get_partition_properties	取得指定之分割區的屬性。屬性字典中的索引鍵包括： eventhub_name str) ( id (str) beginning_sequence_number (int) last_enqueued_sequence_number (int) last_enqueued_offset str) ( last_enqueued_time_utc (UTC datetime.datetime) is_empty (bool)
receive	使用選擇性負載平衡和檢查點，從分割區 () 接收事件。
receive_batch	使用選擇性負載平衡和檢查點，以批次方式接收分割區 () 的事件。

close

停止從事件中樞擷取事件，並關閉基礎 AMQP 連線和連結。

async close() -> None

傳回類型

None

範例

關閉用戶端。


   import os

   event_hub_connection_str = os.environ['EVENT_HUB_CONN_STR']
   eventhub_name = os.environ['EVENT_HUB_NAME']

   from azure.eventhub.aio import EventHubConsumerClient
   consumer = EventHubConsumerClient.from_connection_string(
       conn_str=event_hub_connection_str,
       consumer_group='$Default',
       eventhub_name=eventhub_name  # EventHub name should be specified if it doesn't show up in connection string.
   )

   logger = logging.getLogger("azure.eventhub")

   async def on_event(partition_context, event):
       # Put your code here.
       # If the operation is i/o intensive, async will have better performance.
       logger.info("Received event from partition: {}".format(partition_context.partition_id))

   # The receive method is a coroutine which will be blocking when awaited.
   # It can be executed in an async task for non-blocking behavior, and combined with the 'close' method.

   recv_task = asyncio.ensure_future(consumer.receive(on_event=on_event))
   await asyncio.sleep(3)  # keep receiving for 3 seconds
   recv_task.cancel()  # stop receiving

   # Close down the consumer handler explicitly.
   await consumer.close()

from_connection_string

從連接字串建立 EventHubConsumerClient。

from_connection_string(conn_str: str, consumer_group: str, *, eventhub_name: str | None = None, logging_enable: bool = False, http_proxy: Dict[str, str | int] | None = None, auth_timeout: float = 60, user_agent: str | None = None, retry_total: int = 3, transport_type: TransportType = TransportType.Amqp, checkpoint_store: 'CheckpointStore' | None = None, load_balancing_interval: float = 30, **kwargs: Any) -> EventHubConsumerClient

參數

conn_str: str

必要

事件中樞的連接字串。

consumer_group: str

必要

從這個取用者群組的事件中樞接收事件。

eventhub_name: str

要連接用戶端的特定事件中樞路徑。

logging_enable: bool

是否要將網路追蹤記錄輸出至記錄器。預設值為 False。

http_proxy: dict

HTTP Proxy 設定。這必須是具有下列索引鍵的字典： 'proxy_hostname' ( str 值) 和 'proxy_port' (int 值) 。此外，也可能會出現下列金鑰： 'username'、'password'。

auth_timeout: float

等候服務授權權杖的秒數。預設值為 60 秒。如果設定為 0，則不會從用戶端強制執行逾時。

user_agent: str

如果指定，這會新增到使用者代理程式字串前面。

retry_total: int

retry_backoff_factor: float

retry_backoff_max: float

退班時間上限。預設值為 120 秒， (2 分鐘) 。

retry_mode: str

重試嘗試之間的延遲行為。支援的值為 'fixed' 或 'exponential'，其中預設值為 'exponential'。

idle_timeout: float

transport_type: TransportType

checkpoint_store: Optional[CheckpointStore]

load_balancing_interval: float

負載平衡啟動時。這是兩個負載平衡評估之間的間隔，以秒為單位。預設值為 30 秒。

partition_ownership_expiration_interval: float

load_balancing_strategy: str 或 LoadBalancingStrategy

custom_endpoint_address: Optional[str]

connection_verify: Optional[str]

SSL 憑證之自訂CA_BUNDLE檔案的路徑，用來驗證連線端點的身分識別。預設值為 None，在此情況下會使用 certifi.where () 。

uamqp_transport: bool

是否要使用 uamqp 程式庫作為基礎傳輸。預設值為 False，而純 Python AMQP 程式庫將作為基礎傳輸。

傳回類型

EventHubConsumerClient

範例

從連接字串建立 EventHubConsumerClient 的新實例。


   import os
   from azure.eventhub.aio import EventHubConsumerClient
   event_hub_connection_str = os.environ['EVENT_HUB_CONN_STR']
   eventhub_name = os.environ['EVENT_HUB_NAME']
   consumer = EventHubConsumerClient.from_connection_string(
       conn_str=event_hub_connection_str,
       consumer_group='$Default',
       eventhub_name=eventhub_name  # EventHub name should be specified if it doesn't show up in connection string.
   )

get_eventhub_properties

取得事件中樞的屬性。

傳回字典中的索引鍵包括：

eventhub_name str) (
created_at (UTC datetime.datetime)
partition_ids (list[str])

async get_eventhub_properties() -> Dict[str, Any]

傳回

包含事件中樞相關資訊的字典。

傳回類型

dict

例外狀況

EventHubError

get_partition_ids

取得事件中樞的資料分割識別碼。

async get_partition_ids() -> List[str]

傳回

資料分割識別碼的清單。

傳回類型

list[str]

例外狀況

EventHubError

get_partition_properties

取得指定之分割區的屬性。

屬性字典中的索引鍵包括：

eventhub_name str) (
id (str)
beginning_sequence_number (int)
last_enqueued_sequence_number (int)
last_enqueued_offset str) (
last_enqueued_time_utc (UTC datetime.datetime)
is_empty (bool)

async get_partition_properties(partition_id: str) -> Dict[str, Any]

參數

partition_id: str

必要

目標分割區識別碼。

傳回

包含資料分割屬性的字典。

傳回類型

dict

例外狀況

EventHubError

receive

使用選擇性負載平衡和檢查點，從分割區 () 接收事件。

async receive(on_event: Callable[['PartitionContext', 'EventData' | None], Awaitable[None]], *, max_wait_time: float | None = None, partition_id: str | None = None, owner_level: int | None = None, prefetch: int = 300, track_last_enqueued_event_properties: bool = False, starting_position: str | int | datetime.datetime | Dict[str, Any] | None = None, starting_position_inclusive: bool | Dict[str, bool] = False, on_error: Callable[['PartitionContext', Exception], Awaitable[None]] | None = None, on_partition_initialize: Callable[['PartitionContext'], Awaitable[None]] | None = None, on_partition_close: Callable[['PartitionContext', 'CloseReason'], Awaitable[None]] | None = None) -> None

參數

on_event: Callable[PartitionContext, Optional[EventData]]

必要

用於處理已接收事件的回呼函式。回呼會採用兩個參數： partition_coNtext ，其中包含已接收的事件分割內容和事件。回呼函式的定義方式如下： on_event (partition_coNtext、事件) 。如需詳細的資料分割內容資訊，請參閱 PartitionContext 。

max_wait_time: float

事件處理器在呼叫回呼之前會等候的最大間隔，以秒為單位。如果未在此間隔內收到任何事件，則會以None呼叫on_event回呼。如果此值設定為 None 或 0 (預設) ，在收到事件之前，將不會呼叫回呼。

partition_id: str

如果指定，用戶端只會從這個分割區接收。否則，用戶端會從所有分割區接收。

owner_level: int

獨佔取用者的優先順序。如果已設定owner_level，將會建立獨佔取用者。具有較高owner_level的取用者具有較高的獨佔優先權。擁有者層級也稱為取用者的「Epoch 值」。

prefetch: int

要從服務預先擷取以進行處理的事件數目。預設值為 300。

track_last_enqueued_event_properties: bool

指出取用者是否應該在其相關聯的分割區上要求最後排入佇列事件的資訊，並在收到事件時追蹤該資訊。當追蹤資料分割最後排入佇列事件的相關資訊時，從事件中樞服務收到的每個事件都會包含有關資料分割的中繼資料。這會導致少量額外的網路頻寬耗用量，在考慮使用事件中樞用戶端定期提出分割區屬性的要求時，通常會是一個偏好的取捨。預設會設定為 False 。

starting_position: str, int, datetime 或 dict[str,any]

如果沒有分割區的檢查點資料，請開始從這個事件位置接收。如果有的話，將會使用檢查點資料。這可以是具有分割區識別碼的聽寫，做為個別分割區的值，或是所有分割區的單一值。實數值型別可以是 str、int 或 datetime.datetime。此外，支援從資料流程開頭接收的值「-1」，以及只接收新事件的「@latest」。

starting_position_inclusive: bool 或 dict[str,bool]

判斷指定的starting_position是否包含 (> =) > () 。 True 表示包含，若為 False 則為獨佔。這可以是分割區識別碼為索引鍵的聽寫，而 bool 是值，指出特定分割區的starting_position是否包含。這也可以是所有starting_position的單一 bool 值。預設值是 False。

on_error: Callable[[PartitionContext, Exception]]

在重試嘗試用盡或負載平衡程式期間收到錯誤時，將會呼叫的回呼函式。回呼會採用兩個參數： partition_coNtext ，其中包含分割區資訊和錯誤是例外狀況。 如果在 負載平衡過程中引發錯誤，partition_coNtext可能是 None。回呼的定義應該如下： on_error (partition_coNtext，錯誤) 。如果在on_event回呼期間引發未處理的例外狀況，也會呼叫on_error回呼。

on_partition_initialize: Callable[[PartitionContext]]

在特定分割區的取用者完成初始化之後所呼叫的回呼函式。建立新的內部資料分割取用者以接管失敗和已關閉內部資料分割取用者的接收程式時，也會呼叫它。回呼會採用單一參數： partition_coNtext 其中包含資料分割資訊。回呼的定義應該如下： on_partition_initialize (partition_coNtext) 。

on_partition_close: Callable[[PartitionContext, CloseReason]]

在關閉特定分割區的取用者之後呼叫的回呼函式。當重試嘗試用盡之後，在接收期間引發錯誤時，也會呼叫它。回呼會採用兩個參數： partition_coNtext ，其中包含關閉的資料分割資訊和原因。回呼的定義方式如下： on_partition_close (partition_coNtext，原因) 。如需各種關閉原因，請參閱 CloseReason 。

傳回類型

None

範例

從 EventHub 接收事件。


       logger = logging.getLogger("azure.eventhub")

       async def on_event(partition_context, event):
           # Put your code here.
           # If the operation is i/o intensive, async will have better performance.
           logger.info("Received event from partition: {}".format(partition_context.partition_id))

       async with consumer:
           await consumer.receive(
               on_event=on_event,
               starting_position="-1",  # "-1" is from the beginning of the partition.
           )

receive_batch

使用選擇性負載平衡和檢查點，以批次方式接收分割區 () 的事件。

async receive_batch(on_event_batch: Callable[['PartitionContext', List['EventData']], Awaitable[None]], *, max_batch_size: int = 300, max_wait_time: float | None = None, partition_id: str | None = None, owner_level: int | None = None, prefetch: int = 300, track_last_enqueued_event_properties: bool = False, starting_position: str | int | datetime.datetime | Dict[str, Any] | None = None, starting_position_inclusive: bool | Dict[str, bool] = False, on_error: Callable[['PartitionContext', Exception], Awaitable[None]] | None = None, on_partition_initialize: Callable[['PartitionContext'], Awaitable[None]] | None = None, on_partition_close: Callable[['PartitionContext', 'CloseReason'], Awaitable[None]] | None = None) -> None

參數

on_event_batch: Callable[PartitionContext, List[EventData]]

必要

用於處理已接收事件的批次的回呼函式。回呼會採用兩個參數： partition_coNtext 包含分割區內容和 event_batch，也就是收到的事件。回呼函式的定義方式如下： on_event_batch (partition_coNtext，event_batch) 。如果max_wait_time不是 None 或 0，而且在max_wait_time之後未收到任何事件，event_batch可能是空的清單。如需詳細的資料分割內容資訊，請參閱 PartitionContext 。

max_batch_size: int

批次中傳遞至回呼 on_event_batch的事件數目上限。如果實際收到的事件數目大於 max_batch_size，則收到的事件會分割成批次，並呼叫每個批次的回呼，最多 max_batch_size 個事件。

max_wait_time: float

事件處理器在呼叫回呼之前會等候的最大間隔，以秒為單位。如果未在此間隔內收到任何事件，則會以空清單呼叫 on_event_batch 回呼。如果此值設定為 None 或 0 (預設) ，在收到事件之前，將不會呼叫回呼。

partition_id: str

如果指定，用戶端只會從這個分割區接收。否則，用戶端會從所有分割區接收。

owner_level: int

prefetch: int

要從服務預先擷取以進行處理的事件數目。預設值為 300。

track_last_enqueued_event_properties: bool

starting_position: str, int, datetime 或 dict[str,any]

starting_position_inclusive: bool 或 dict[str,bool]

on_error: Callable[[PartitionContext, Exception]]

on_partition_initialize: Callable[[PartitionContext]]

on_partition_close: Callable[[PartitionContext, CloseReason]]

傳回類型

None

範例

從 EventHub 批次接收事件。


       logger = logging.getLogger("azure.eventhub")

       async def on_event_batch(partition_context, event_batch):
           # Put your code here.
           # If the operation is i/o intensive, async will have better performance.
           logger.info(
               "{} events received from partition: {}".format(len(event_batch), partition_context.partition_id)
           )

       async with consumer:
           await consumer.receive_batch(
               on_event_batch=on_event_batch,
               starting_position="-1",  # "-1" is from the beginning of the partition.
           )

Share via

EventHubConsumerClient 類別

建構函式

參數

範例

方法

close

傳回類型

範例

from_connection_string

參數

傳回類型

範例

get_eventhub_properties

傳回

傳回類型

例外狀況

get_partition_ids

傳回

傳回類型

例外狀況

get_partition_properties

參數

傳回

傳回類型

例外狀況

receive

參數

傳回類型

範例

receive_batch

參數

傳回類型

範例

意見反應

其他資源