如何使用Python从Azure IoT Hub消费并存储设备数据？

别纠结啦，其实IoT Hub和Event Hub本来就打通了——IoT Hub内置了Event Hub兼容的内置端点，所以你完全可以用熟悉的Azure Event Hubs Python SDK来消费设备消息！下面给你一步步拆解操作：

1. 获取IoT Hub的Event Hub兼容信息 #

首先从Azure门户里拿到几个关键参数：

• 登录Azure门户，进入你的IoT Hub资源
• 左侧菜单找到内置终结点，点击进入
• 复制以下信息：
  • 事件中心兼容终结点：类似Event Hub的连接地址
  • 事件中心兼容名称：相当于Event Hub的实例名称
  • 同时获取IoT Hub的访问密钥：在共享访问策略里选择iothubowner（或自定义的具备service connect权限的策略），复制其连接字符串

2. 安装必要的Python包 #

打开终端，安装核心SDK和可选的检查点存储包（用于持久化消费进度）：

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

• azure-eventhub：核心的Event Hub消费SDK
• azure-eventhub-checkpointstoreblob-aio：用Azure Blob存储保存消费检查点，避免重启后重复处理消息

3. 编写Python消费代码 #

下面是一个适合生产环境的异步消费示例，自带检查点持久化功能：

from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
import asyncio

# 自定义消息处理回调函数
async def process_device_message(partition_context, event):
    # 提取设备元数据与消息内容
    device_id = event.system_properties["iothub-connection-device-id"]
    message_body = event.body_as_str()
    enqueue_time = event.system_properties["iothub-enqueuedtime"]

    print(f"[{enqueue_time}] 收到设备 {device_id} 的消息: {message_body}")

    # 在这里添加你的数据库写入逻辑，比如写入SQL Server、Cosmos DB等
    # 示例（伪代码）：
    # await db_client.insert_record(
    #     device_id=device_id,
    #     content=message_body,
    #     received_time=enqueue_time
    # )

    # 更新检查点，记录当前消费进度
    await partition_context.update_checkpoint(event)

async def main():
    # 替换成你自己的资源配置
    IOTHUB_CONNECTION_STRING = "你的IoT Hub连接字符串"
    EVENT_HUB_COMPATIBLE_NAME = "你的事件中心兼容名称"
    STORAGE_CONNECTION_STRING = "你的Azure存储账户连接字符串"
    BLOB_CONTAINER_NAME = "你的存储容器名称（用于保存检查点）"

    # 初始化检查点存储
    checkpoint_store = BlobCheckpointStore.from_connection_string(
        STORAGE_CONNECTION_STRING, BLOB_CONTAINER_NAME
    )

    # 创建Event Hub消费者客户端
    consumer_client = EventHubConsumerClient.from_connection_string(
        conn_str=IOTHUB_CONNECTION_STRING,
        consumer_group="$Default",  # 默认消费组，可根据业务自定义
        eventhub_name=EVENT_HUB_COMPATIBLE_NAME,
        checkpoint_store=checkpoint_store
    )

    # 持续监听并消费消息
    async with consumer_client:
        await consumer_client.receive(
            on_event=process_device_message,
            starting_position="-1"  # "-1"从最早未消费消息开始，"@latest"从最新消息开始
        )

if __name__ == "__main__":
    asyncio.run(main())

关键细节说明 #

• 消费组：IoT Hub支持多消费组，每个组独立维护消费进度，适合不同业务场景并行处理消息
• 系统属性：通过event.system_properties可以获取设备ID、消息入队时间等关键元数据，常用键包括iothub-connection-device-id、iothub-enqueuedtime
• 检查点：生产环境建议开启检查点，避免程序重启后重复消费；测试场景可省略检查点，直接使用无状态消费
• 数据库适配：消息处理函数里可以根据你的数据库类型（MySQL、PostgreSQL等）编写对应的写入逻辑，SDK不限制数据库选型

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