如何使用azure-event-hubs-python从多个分区获取事件？

你遇到的问题根源很明确：run()方法是阻塞式的——当你在循环里调用它时，第一个分区（分区0）的接收器启动后会直接占用主线程，后面的循环迭代根本没机会执行，自然只有分区0在工作。

下面给你两种可行的解决方案，分别对应旧版和新版SDK的场景：

针对你当前使用的旧版SDK（azure-eventhub v4及以下） #

我们可以用Python的threading模块，把每个分区接收器的运行逻辑放到独立线程中，避免主线程被阻塞。

示例代码如下：

import threading
from azure.eventhub import EventHubClient, Receiver

# 替换成你的Event Hub配置
ADDRESS = "你的Event Hub地址"
CONSUMER_GROUP = "$Default"
OFFSET = "@latest"

# 你的自定义接收器类（假设已经实现了必要的回调）
class MyReceiver(Receiver):
    def on_event(self, event):
        print(f"来自分区 {self.partition_id} 的消息: {event.body_as_str()}")
    
    def on_error(self, error):
        print(f"分区 {self.partition_id} 出错: {str(error)}")
    
    def on_close(self, reason):
        print(f"分区 {self.partition_id} 的接收器已关闭: {reason}")

# 用线程列表管理所有分区接收器
thread_list = []

for partition_id in range(4):
    # 为每个分区创建独立的客户端和接收器
    client = EventHubClient(ADDRESS)
    receiver = client.subscribe(
        MyReceiver(str(partition_id)),
        consumer_group=CONSUMER_GROUP,
        partition=str(partition_id),
        offset=OFFSET
    )
    # 将接收器的run方法放到线程中启动
    receiver_thread = threading.Thread(target=receiver.run)
    thread_list.append(receiver_thread)
    receiver_thread.start()

# 可选：让主线程等待所有接收器线程结束（避免程序直接退出）
for thread in thread_list:
    thread.join()

关键说明： #

• 每个分区的接收器在独立线程中运行，互不阻塞，所有分区都能正常监听；
• 确保你的Event Hub实际配置了至少4个分区（可以在Azure门户的Event Hub详情页查看），否则会抛出分区不存在的错误；
• 同一个消费者组下，每个分区只能有一个活跃的接收器（除非你启用了负载均衡，但这里我们是一对一绑定分区，所以没问题）。

如果你升级到新版SDK（azure-eventhub v5及以上） #

新版SDK大幅简化了多分区监听的逻辑，不需要手动管理线程，EventHubConsumerClient会自动为每个分区创建接收器并分配资源。

示例代码：

from azure.eventhub import EventHubConsumerClient

# 替换成你的配置
CONNECTION_STR = "你的Event Hub连接字符串"
EVENT_HUB_NAME = "你的Event Hub名称"
CONSUMER_GROUP = "$Default"

def process_event_batch(partition_context, events):
    """批量处理来自某个分区的消息"""
    print(f"正在处理分区 {partition_context.partition_id} 的 {len(events)} 条消息")
    for event in events:
        print(f"消息内容: {event.body_as_str()}")
    # 处理完成后可以更新检查点（下次重启会从检查点继续消费）
    partition_context.update_checkpoint()

# 创建消费者客户端
client = EventHubConsumerClient.from_connection_string(
    conn_str=CONNECTION_STR,
    consumer_group=CONSUMER_GROUP,
    eventhub_name=EVENT_HUB_NAME
)

# 启动多分区监听，默认会监听所有分区
with client:
    client.receive_batch(
        on_event_batch=process_event_batch,
        starting_position="-1"  # 从最早的未消费消息开始，也可以用"@latest"从最新消息开始
    )

新版优势： #

• 自动管理分区和线程，无需手动编写线程逻辑；
• 内置检查点机制，支持断点续传；
• 更好的性能和错误处理能力。

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