Azure Event Hub多分区Python订阅重复消费问题求助

针对你遇到的同一消息被多台VM上的接收端重复消费的问题，可按以下步骤逐一排查：

1. 确认起始位置配置的影响 #

你的代码中指定了starting_position="-1"，该参数表示从分区最开始消费，但需注意：

• 当使用Checkpoint Store时，仅在该分区无已有Checkpoint时，此参数才会生效；如果已有Checkpoint，客户端应从Checkpoint记录的位置继续消费。
• 若客户端启动时，对应分区的Checkpoint尚未被写入（比如第一个客户端刚启动还没处理任何消息），后续启动的客户端会因无Checkpoint而从分区开头读取，导致重复消费。
• 建议将starting_position改为"@latest"（从最新消息开始消费），或者移除该参数使用默认值，避免在已有Checkpoint的情况下强制从头读取。

修改后的receive调用示例：

await client.receive(on_event=on_event)  # 不指定starting_position，优先使用Checkpoint

2. 验证Checkpoint的读写状态 #

手动检查Blob存储中的Checkpoint #

登录Azure门户，进入你的Blob存储账户，查看指定容器下的路径：
eventhubs/<你的Event Hub名称>/$Default/<分区ID>/checkpoint

• 确认每个分区都存在对应的Checkpoint Blob
• 打开Blob内容，检查是否包含offset和sequence_number字段，且数值随消息处理逐步递增

在代码中添加日志验证 #

修改on_event和main函数，添加日志输出Checkpoint相关信息：

async def on_event(partition_context, event):
    # 打印当前事件的offset和分区ID
    print(f"处理事件: 分区={partition_context.partition_id}, Offset={event.offset}, 内容={event.body_as_str(encoding='UTF-8')}")
    try:
        await partition_context.update_checkpoint(event)
        # 打印更新后的Checkpoint位置
        checkpoint = await partition_context.get_checkpoint()
        print(f"已更新Checkpoint: 分区={partition_context.partition_id}, Offset={checkpoint['offset']}")
    except Exception as e:
        print(f"更新Checkpoint失败: {str(e)}")

async def main():
    checkpoint_store = BlobCheckpointStore.from_connection_string(
        BLOB_STORAGE_CONNECTION_STRING, BLOB_CONTAINER_NAME
    )
    # 预加载并打印当前Checkpoint状态
    eventhub_properties = await checkpoint_store.get_eventhub_properties(EVENT_HUB_NAME)
    partitions = eventhub_properties["partition_ids"]
    for partition_id in partitions:
        checkpoint = await checkpoint_store.get_checkpoint(EVENT_HUB_NAME, "$Default", partition_id)
        print(f"初始Checkpoint: 分区={partition_id}, Offset={checkpoint['offset'] if checkpoint else '无'}")
    
    # 后续客户端创建和接收逻辑不变
    client = EventHubConsumerClient.from_connection_string(
        EVENT_HUB_CONNECTION_STR,
        consumer_group="$Default",
        eventhub_name=EVENT_HUB_NAME,
        checkpoint_store=checkpoint_store,
    )
    async with client:
        await client.receive(on_event=on_event)

3. 确认消费组的一致性 #

确保所有5台VM上的客户端都使用完全相同的消费组名称：

• 消费组名称区分大小写，若某台客户端误写为$default（小写），会被视为独立的消费组，导致重复消费
• 检查代码中consumer_group="$Default"的拼写是否完全一致

4. 验证包版本兼容性 #

当前使用的azure-eventhub 5.10.1和azure-eventhub-checkpointstoreblob-aio 1.1.4存在版本匹配问题：

• 官方推荐Checkpoint Store包的版本与Event Hub包版本保持兼容，建议升级到最新稳定版：

  pip install --upgrade azure-eventhub azure-eventhub-checkpointstoreblob-aio
  

• 升级后可修复旧版本中可能存在的Checkpoint同步、负载均衡等已知bug

5. 检查Blob存储的访问权限与网络 #

• 确认VM所在环境能正常访问Blob存储：无VNet防火墙限制、网络连通性正常
• 验证Blob存储连接字符串的正确性：包含有效的账户名和密钥，且该密钥拥有容器的读写权限
• 若使用托管身份，需确认VM的托管身份已被授予Blob存储的Storage Blob Data Contributor角色

6. 检查异步代码的异常处理 #

异步代码中若未捕获异常，可能导致update_checkpoint未被执行：

• 在on_event函数中添加完整的异常捕获，确保即使事件处理失败，也能排查问题，且必要时重试Checkpoint更新

7. 验证分区负载均衡状态 #

Event Hub的消费组会自动将分区分配给不同的客户端，可在代码中打印客户端分配到的分区：

async def on_partition_initialize(partition_context):
    print(f"客户端已分配分区: {partition_context.partition_id}")

async def main():
    # ... 其他代码
    async with client:
        await client.receive(
            on_event=on_event,
            on_partition_initialize=on_partition_initialize
        )

• 查看所有客户端的日志，确认是否存在多个客户端分配到同一分区的情况，若存在则说明负载均衡逻辑异常，需检查网络连通性或升级包版本

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