使用EventHubConsumerClient实现异步Azure Event Hub触发函数遇运行异常

你的代码核心问题是同时混用了Azure Functions Event Hub触发器和原生的EventHubConsumerClient，这两者完全不能同时使用：

• Functions的Event Hub触发器已经由Azure Functions Runtime负责Event Hub的连接、事件拉取和分发，你不需要自己再实例化EventHubConsumerClient
• 你在main函数里启动了一个持续阻塞的client.receive()调用，再加上asyncio.run()在Functions runtime的事件循环中嵌套运行，直接导致进程卡死
• cardinality: "one"配置意味着每次触发只传递单个事件，但你的代码却在尝试持续监听所有分区的事件，逻辑完全冲突

根据你的需求（异步、Blob存储检查点、转发到第二个Event Hub），推荐两种合规的实现方式：

方式一：基于Azure Functions Event Hub触发器（生产环境推荐） #

利用Functions触发器的原生能力，结合异步处理，手动管理检查点到Blob存储，同时完成事件转发。

修正后的__init__.py代码 #

import logging
import os
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData
from azure.storage.blob.aio import BlobServiceClient
import azure.functions as func

# 目标Event Hub配置
TARGET_EVENTHUB_CONN_STR = os.environ.get("TARGET_EVENT_HUB_CONN_STR")
TARGET_EVENTHUB_NAME = os.environ.get("TARGET_EVENT_HUB_NAME")

# Blob检查点存储配置
STORAGE_CONN_STR = os.environ.get("AZURE_STORAGE_CONN_STR")
CHECKPOINT_CONTAINER = os.environ.get("AZURE_STORAGE_NAME")

async def update_checkpoint(partition_id, offset, sequence_number):
    """手动更新检查点到Blob存储"""
    blob_service_client = BlobServiceClient.from_connection_string(STORAGE_CONN_STR)
    async with blob_service_client:
        container_client = blob_service_client.get_container_client(CHECKPOINT_CONTAINER)
        await container_client.create_container(exists_ok=True)
        blob_client = container_client.get_blob_client(f"checkpoint/{partition_id}.json")
        checkpoint_data = {
            "offset": offset,
            "sequence_number": sequence_number
        }
        await blob_client.upload_blob(str(checkpoint_data), overwrite=True)

async def forward_event(event_data):
    """转发事件到目标Event Hub"""
    producer_client = EventHubProducerClient.from_connection_string(
        TARGET_EVENTHUB_CONN_STR,
        eventhub_name=TARGET_EVENTHUB_NAME
    )
    async with producer_client:
        async with producer_client.create_batch() as batch:
            batch.add(EventData(event_data))
            await producer_client.send_batch(batch)

async def main(events: func.EventHubEvent):
    # 批量处理事件（建议用many cardinality提高效率）
    for event in events:
        event_body = event.get_body().decode("UTF-8")
        partition_id = event.metadata["partition_id"]
        logging.info(f"Received event: {event_body} from partition {partition_id}")
        
        # 转发事件到目标Event Hub
        await forward_event(event_body)
        
        # 更新检查点到Blob存储
        await update_checkpoint(
            partition_id,
            event.offset,
            event.sequence_number
        )

对应的function.json配置 #

修改cardinality为many，提升批量处理效率：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "events",
      "direction": "in",
      "eventHubName": "<My_event_hub_name>",
      "connection": "<My_event_hub_co_str>",
      "cardinality": "many",
      "consumerGroup": "$Default"
    }
  ]
}

方式二：直接使用EventHubConsumerClient（无Functions触发器） #

如果你更倾向于用原生SDK的消费逻辑，需要完全移除Event Hub触发器，改用Timer触发器启动一次后持续运行（注意Functions的冷启动特性）：

修正后的__init__.py代码 #

import logging
import asyncio
import os
from azure.eventhub.aio import EventHubConsumerClient, EventHubProducerClient
from azure.eventhub import EventData
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
import azure.functions as func

# 源Event Hub配置
SOURCE_CONN_STR = os.environ.get("EVENT_HUB_CONN_STR")
SOURCE_EVENTHUB_NAME = os.environ.get("EVENT_HUB_NAME")

# 目标Event Hub配置
TARGET_CONN_STR = os.environ.get("TARGET_EVENT_HUB_CONN_STR")
TARGET_EVENTHUB_NAME = os.environ.get("TARGET_EVENT_HUB_NAME")

# Blob检查点存储配置
STORAGE_CONN_STR = os.environ.get("AZURE_STORAGE_CONN_STR")
BLOB_CONTAINER_NAME = os.environ.get("AZURE_STORAGE_NAME")

async def on_event(partition_context, event):
    event_body = event.body_as_str(encoding="UTF-8")
    logging.info(f"Received event: {event_body} from partition {partition_context.partition_id}")
    
    # 转发事件到目标Event Hub
    async with EventHubProducerClient.from_connection_string(
        TARGET_CONN_STR, eventhub_name=TARGET_EVENTHUB_NAME
    ) as producer:
        async with producer.create_batch() as batch:
            batch.add(EventData(event_body))
            await producer.send_batch(batch)
    
    # 更新检查点
    await partition_context.update_checkpoint(event)

async def run_consumer():
    checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONN_STR, BLOB_CONTAINER_NAME)
    client = EventHubConsumerClient.from_connection_string(
        SOURCE_CONN_STR,
        consumer_group="$Default",
        eventhub_name=SOURCE_EVENTHUB_NAME,
        checkpoint_store=checkpoint_store,
    )
    async with client:
        await client.receive(on_event=on_event, starting_position="-1")

# 用Timer触发器启动一次，然后持续运行消费者
async def main(mytimer: func.TimerRequest):
    if mytimer.past_due:
        logging.info('The timer is past due!')
    
    await run_consumer()

对应的function.json配置（Timer触发器） #

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 */5 * * * *"
    }
  ]
}

• 绝对不要在Functions触发器中嵌套asyncio.run()，Functions runtime已经维护了自己的事件循环
• 使用方式一时，检查点的路径和格式可根据需求调整，确保每个分区的检查点唯一
• 生产环境优先用cardinality: "many"批量处理事件，减少函数触发次数、降低成本
• 确保依赖包（azure-functions, azure-eventhub, azure-storage-blob）版本兼容，推荐使用最新稳定版

内容的提问来源于stack exchange，提问作者BetraVageur